From 36d98bf7467ffd51c28c04dc08ebb184d38be76a Mon Sep 17 00:00:00 2001 From: Upper9527 Date: Wed, 21 Oct 2026 11:48:36 +0800 Subject: [PATCH] Stream chat deltas live with cancellation and complete-turn tracking --- README.md | 6 ++ assets/README_ZH.md | 6 ++ skillclaw/api_server.py | 64 +++++++++++- skillclaw/chat_stream.py | 54 ++++++++++ skillclaw/chat_upstream.py | 8 +- tests/test_realtime_chat.py | 197 ++++++++++++++++++++++++++++++++++++ 6 files changed, 329 insertions(+), 6 deletions(-) create mode 100644 skillclaw/chat_stream.py create mode 100644 tests/test_realtime_chat.py diff --git a/README.md b/README.md index 753aaed..7d87bc7 100644 --- a/README.md +++ b/README.md @@ -541,6 +541,12 @@ On a connection failure or HTTP 408/429/500/502/503/504, the proxy tries the nex The entire operation has six attempts and the existing `SKILLCLAW_LLM_REQUEST_TIMEOUT_S` time budget, including backoff and `Retry-After`. Long final quota windows fail promptly. Native Responses and Bedrock forwarding, and the standalone evolve-server LLM client, do not use this list. Restart the proxy after editing configuration. +### Real-time Chat streaming + +OpenAI-compatible `/v1/chat/completions` now forwards `stream: true` incrementally, including separate `reasoning_content` and structured tool-call deltas. The completed turn is recorded once; cancellation or incomplete/error streams close the upstream and do not record a successful turn. Failover is allowed before a successful upstream stream starts, never after emitted output. Errors after HTTP headers are sent appear as an SSE `error` object without `[DONE]`. + +Streaming requires structured OpenAI tool calls; textual tool-tag repair is available only for buffered requests. Bedrock and Chat-to-Anthropic/Responses conversion retain their existing behavior. Disable buffering in any reverse proxy in front of SkillClaw. + ## Quick Reference **Operational model** — Every user has their own `skillclaw` process and `~/.skillclaw/config.yaml`. Each group maps to a `group_id`; one evolve server watches that namespace, turns sessions into skills, and writes them back. Joining a group only requires the correct sharing config on the client side. diff --git a/assets/README_ZH.md b/assets/README_ZH.md index 550c572..2ffc6cf 100644 --- a/assets/README_ZH.md +++ b/assets/README_ZH.md @@ -491,6 +491,12 @@ llm: 整个请求最多尝试六次，并共享现有 `SKILLCLAW_LLM_REQUEST_TIMEOUT_S` 总时限（包括退避和 Retry-After）。最后一个端点的长配额窗口会快速失败。原生 Responses、Bedrock 和独立 evolve-server 的 LLM 调用不使用该列表。修改后重启代理。 +### Chat 实时流式输出 + +OpenAI-compatible `/v1/chat/completions` 的 `stream: true` 现在会实时转发增量，包括独立的 `reasoning_content` 和结构化工具调用分片。完成后只记录一次会话回合；客户端取消、上游错误或不完整流会关闭上游连接，不记录成功回合。仅在成功开启上游流之前允许故障切换，输出后不重放请求。HTTP 响应头发出后的错误以 SSE `error` 对象返回，不发送 `[DONE]`。 + +流式请求需要上游提供 OpenAI 结构化工具调用；文本工具标签修复仅用于非流式请求。Bedrock、Chat 转 Anthropic/Responses 路径维持原有行为。SkillClaw 前面若有反向代理，需关闭响应缓冲。 + ## 快速参考 **运行关系** — 每个用户有自己的 `skillclaw` 进程和 `~/.skillclaw/config.yaml`；每个群组对应一个 `group_id`，由一个 Evolver 持续读取会话、产出技能、写回存储。加入群组只需正确配置共享存储，无需部署服务端。 diff --git a/skillclaw/api_server.py b/skillclaw/api_server.py index 878d546..d07db96 100644 --- a/skillclaw/api_server.py +++ b/skillclaw/api_server.py @@ -745,6 +745,7 @@ def _assemble_streaming_chat_completion( { "role": "assistant", "content_parts": [], + "reasoning_parts": [], "tool_calls": {}, "finish_reason": None, }, @@ -753,6 +754,8 @@ def _assemble_streaming_chat_completion( if isinstance(delta.get("role"), str): entry["role"] = delta["role"] + if isinstance(delta.get("reasoning_content"), str): + entry["reasoning_parts"].append(delta["reasoning_content"]) content = delta.get("content") if isinstance(content, str): entry["content_parts"].append(content) @@ -794,6 +797,8 @@ def _assemble_streaming_chat_completion( "role": entry["role"], "content": "".join(entry["content_parts"]), } + if entry["reasoning_parts"]: + message["reasoning_content"] = "".join(entry["reasoning_parts"]) if entry["tool_calls"]: message["tool_calls"] = [entry["tool_calls"][i] for i in sorted(entry["tool_calls"])] choices.append( @@ -1604,6 +1609,14 @@ async def chat_completions( # X-Session-Done or body session_done trigger session close. stream = bool(body.get("stream", False)) + if stream and owner.config.llm_provider != "bedrock": + if not isinstance(body.get("messages"), list) or not body["messages"]: + raise HTTPException(status_code=400, detail="messages must be a non-empty list") + return StreamingResponse( + owner._stream_chat_and_track(body, session_id, turn_type, session_done), + media_type="text/event-stream", + headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"}, + ) result = await owner._handle_request( body, session_id=session_id, @@ -2313,6 +2326,7 @@ async def _handle_request( session_id: str, turn_type: str, session_done: bool, + forwarder=None, ) -> dict[str, Any]: protocol = str(body.get("_skillclaw_protocol") or "").strip() messages = body.get("messages") @@ -2396,12 +2410,13 @@ def _prompt_len(msgs): forward_body = {k: v for k, v in body.items() if k not in _NON_STANDARD_BODY_KEYS} forward_body["stream"] = False - forward_body.pop("stream_options", None) + if not forwarder: + forward_body.pop("stream_options", None) if "model" not in forward_body: forward_body["model"] = self._served_model forward_body["messages"] = messages # potentially skill-injected - output = await self._forward_to_llm(forward_body) + output = await (forwarder or self._forward_to_llm)(forward_body) output["model"] = forward_body.get("model") or self._served_model choice = output.get("choices", [{}])[0] @@ -2422,7 +2437,7 @@ def _prompt_len(msgs): # Upstream models sometimes emit tool calls as text tags instead of # structured `message.tool_calls`; parse and normalize both sources. - clean_content, text_tool_calls = _extract_tool_calls_from_text(content) + clean_content, text_tool_calls = (content, []) if forwarder else _extract_tool_calls_from_text(content) if text_tool_calls: content = clean_content tool_calls.extend(text_tool_calls) @@ -2543,6 +2558,49 @@ def _prompt_len(msgs): # LLM forwarding # # ------------------------------------------------------------------ # + async def _stream_chat_and_track(self, body, session_id, turn_type, session_done): + from .chat_stream import chat_events + from .chat_upstream import forward_chat + + queue = asyncio.Queue(maxsize=16) + + async def forward(prepared): + async def collect(response, model): + events = [] + async for event in chat_events(response): + events.append(event) + outgoing = dict(event, model=body.get("model") or self._served_model, session_id=session_id) + await queue.put(outgoing) + return _assemble_streaming_chat_completion(events, fallback_model=model) + + return await forward_chat( + self.config, prepared, timeout=_llm_request_timeout_seconds(), collect_stream=collect, stream=True + ) + + async def produce(): + try: + await self._handle_request(body, session_id, turn_type, session_done, forwarder=forward) + except Exception as exc: + status = exc.status_code if isinstance(exc, HTTPException) else 500 + await queue.put({"error": {"message": "Chat stream failed", "type": "upstream_error", "code": status}}) + await queue.put(False) + else: + await queue.put(True) + + producer = asyncio.create_task(produce()) + try: + while True: + item = await queue.get() + if isinstance(item, bool): + if item: + yield "data: [DONE]\n\n" + break + yield "data: " + json.dumps(item, ensure_ascii=False) + "\n\n" + finally: + if not producer.done(): + producer.cancel() + await asyncio.gather(producer, return_exceptions=True) + async def _forward_to_llm(self, body: dict[str, Any]) -> dict[str, Any]: """Forward to a real LLM API. diff --git a/skillclaw/chat_stream.py b/skillclaw/chat_stream.py new file mode 100644 index 0000000..c7c14a2 --- /dev/null +++ b/skillclaw/chat_stream.py @@ -0,0 +1,54 @@ +"""Validate SSE frames while preserving incremental chat deltas.""" + +from __future__ import annotations + +import json + +from fastapi import HTTPException + + +async def chat_events(response): + data = [] + finished = {} + + def parse(): + payload = "\n".join(data) + data.clear() + if payload == "[DONE]": + if not finished or not all(finished.values()): + raise ValueError("Incomplete chat stream") + return None + event = json.loads(payload) + if not isinstance(event, dict) or event.get("error"): + raise ValueError("Upstream stream error") + choices = event.get("choices") + if not isinstance(choices, list): + raise ValueError("Invalid chat stream choices") + for choice in choices: + if not isinstance(choice, dict) or not isinstance(choice.get("delta", {}), dict): + raise ValueError("Invalid chat stream delta") + index = int(choice.get("index", 0)) + finished[index] = finished.get(index, False) or bool(choice.get("finish_reason")) + return event + + try: + async for line in response.aiter_lines(): + if line.startswith("data:"): + data.append(line[5:].lstrip(" ")) + elif not line and data: + event = parse() + if event is None: + return + yield event + if data: + event = parse() + if event is None: + return + yield event + # Even a finish_reason frame can precede a transport failure. Require + # the protocol terminator before committing this turn to session state. + raise ValueError("Missing stream terminator") + except Exception as exc: + # In particular, do not expose a TransportError to failover after + # successful response headers: emitted text must never be replayed. + raise HTTPException(status_code=502, detail="Upstream chat stream failed or ended incompletely") from exc diff --git a/skillclaw/chat_upstream.py b/skillclaw/chat_upstream.py index 9715fba..035585e 100644 --- a/skillclaw/chat_upstream.py +++ b/skillclaw/chat_upstream.py @@ -65,6 +65,8 @@ def chat_endpoints(config, requested_model): def prepare_chat_request(config, body, endpoint, *, stream=False): payload = copy.deepcopy({k: v for k, v in body.items() if k not in {"logprobs", "top_logprobs", "stream_options"}}) payload.update(model=endpoint["model_id"], stream=stream) + if stream and isinstance(body.get("stream_options"), dict): + payload["stream_options"] = copy.deepcopy(body["stream_options"]) headers = {} if endpoint["api_key"]: headers["Authorization"] = f"Bearer {endpoint['api_key']}" @@ -97,12 +99,12 @@ def retry_after_seconds(response): return None -async def forward_chat(config, body, *, timeout, collect_stream): +async def forward_chat(config, body, *, timeout, collect_stream, stream=False): endpoints = chat_endpoints(config, body.get("model", "")) async def run(): index = 0 - url, payload, headers = prepare_chat_request(config, body, endpoints[index]) + url, payload, headers = prepare_chat_request(config, body, endpoints[index], stream=stream) last_status = None async with httpx.AsyncClient(timeout=timeout) as client: # A shared budget bounds the total attempts, including compatibility @@ -150,7 +152,7 @@ async def run(): raise HTTPException(status_code=502, detail="Upstream LLM returned invalid JSON") from exc if index + 1 < len(endpoints): index += 1 - url, payload, headers = prepare_chat_request(config, body, endpoints[index]) + url, payload, headers = prepare_chat_request(config, body, endpoints[index], stream=stream) continue if attempt < 5: await asyncio.sleep(delay if delay is not None else min(2**attempt + random.uniform(0, 1), 30)) diff --git a/tests/test_realtime_chat.py b/tests/test_realtime_chat.py new file mode 100644 index 0000000..0c4a72f --- /dev/null +++ b/tests/test_realtime_chat.py @@ -0,0 +1,197 @@ +import asyncio +import json + +import httpx +import pytest + +from skillclaw.api_server import SkillClawAPIServer +from skillclaw.config import SkillClawConfig + + +def chunk(delta=None, finish=None): + return ( + "data: " + + json.dumps({"id": "chat-1", "choices": [{"index": 0, "delta": delta or {}, "finish_reason": finish}]}) + + "\n\n" + ).encode() + + +class GatedStream(httpx.AsyncByteStream): + def __init__(self, tail=None, error=False): + self.release = asyncio.Event() + self.closed = False + self.error = error + self.tail = tail or [chunk({"content": " world"}, "stop"), b"data: [DONE]\n\n"] + + async def __aiter__(self): + yield chunk({"role": "assistant", "content": "hello", "reasoning_content": "separate"}) + await self.release.wait() + if self.error: + raise httpx.ReadError("private upstream details") + for part in self.tail: + yield part + + async def aclose(self): + self.closed = True + + +def setup(monkeypatch, tmp_path, stream, first_status=None): + server = SkillClawAPIServer( + SkillClawConfig( + llm_api_base="http://primary.test/v1", + llm_model_id="primary", + llm_fallbacks=[{"api_base": "http://backup.test/v1", "model_id": "backup"}], + claw_type="none", + use_skills=False, + use_prm=False, + record_enabled=False, + record_dir=str(tmp_path), + ) + ) + requests = [] + + def handler(request): + requests.append(request) + assert json.loads(request.content)["stream"] is True + if len(requests) == 1 and first_status: + return httpx.Response(first_status, json={"error": "unavailable"}) + return httpx.Response(200, stream=stream) + + client_cls = httpx.AsyncClient + monkeypatch.setattr(httpx, "AsyncClient", lambda **kw: client_cls(transport=httpx.MockTransport(handler), **kw)) + generator = server._stream_chat_and_track( + {"model": "public-alias", "messages": [{"role": "user", "content": "hi"}], "stream": True}, + "session-1", + "main", + False, + ) + return server, generator, requests + + +@pytest.mark.asyncio +async def test_first_delta_arrives_before_generation_finishes_and_records_once(monkeypatch, tmp_path): + upstream = GatedStream() + server, stream, requests = setup(monkeypatch, tmp_path, upstream) + first = await asyncio.wait_for(anext(stream), 1) + payload = json.loads(first.removeprefix("data: ")) + assert payload["choices"][0]["delta"]["content"] == "hello" + assert payload["choices"][0]["delta"]["reasoning_content"] == "separate" + assert payload["model"] == "public-alias" + assert server._session_turns.get("session-1", []) == [] + upstream.release.set() + rest = [part async for part in stream] + assert rest[-1] == "data: [DONE]\n\n" + turns = server._session_turns["session-1"] + assert len(turns) == 1 + assert turns[0]["response_text"] == "hello world" + assert turns[0]["reasoning_content"] == "separate" + assert len(requests) == 1 and upstream.closed + + +@pytest.mark.asyncio +async def test_tool_call_fragments_are_preserved_and_assembled_once(monkeypatch, tmp_path): + upstream = GatedStream( + tail=[ + chunk( + { + "tool_calls": [ + { + "index": 0, + "id": "call-a", + "type": "function", + "function": {"name": "read", "arguments": '{"path":'}, + } + ] + } + ), + chunk({"tool_calls": [{"index": 0, "function": {"arguments": '"a"}'}}]}, "tool_calls"), + b"data: [DONE]\n\n", + ] + ) + upstream.release.set() + server, stream, _ = setup(monkeypatch, tmp_path, upstream) + parts = [part async for part in stream] + assert len(parts) == 4 + call = server._session_turns["session-1"][0]["tool_calls"][0] + assert call["id"] == "call-a" + assert call["function"] == {"name": "read", "arguments": '{"path":"a"}'} + + +@pytest.mark.asyncio +async def test_cancellation_closes_upstream_without_recording(monkeypatch, tmp_path): + upstream = GatedStream() + server, stream, requests = setup(monkeypatch, tmp_path, upstream) + await asyncio.wait_for(anext(stream), 1) + await stream.aclose() + assert upstream.closed and len(requests) == 1 + assert server._session_turns.get("session-1", []) == [] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("kind", ["transport", "missing_done", "invalid_json", "error_event"]) +async def test_midstream_error_never_replays_or_records_success(monkeypatch, tmp_path, kind): + tails = { + "missing_done": [chunk({"content": "truncated"})], + "invalid_json": [b"data: invalid\n\n"], + "error_event": [b'data: {"error":{"message":"secret"}}\n\n'], + } + upstream = GatedStream(tail=tails.get(kind), error=kind == "transport") + upstream.release.set() + server, stream, requests = setup(monkeypatch, tmp_path, upstream) + parts = [part async for part in stream] + assert any('"error"' in part for part in parts) + assert not any("[DONE]" in part or "secret" in part or "private upstream" in part for part in parts) + assert server._session_turns.get("session-1", []) == [] + assert len(requests) == 1 and upstream.closed + + +@pytest.mark.asyncio +async def test_failover_before_stream_starts(monkeypatch, tmp_path): + upstream = GatedStream() + upstream.release.set() + server, stream, requests = setup(monkeypatch, tmp_path, upstream, first_status=503) + parts = [part async for part in stream] + assert parts[-1] == "data: [DONE]\n\n" + assert [r.url.host for r in requests] == ["primary.test", "backup.test"] + assert len(server._session_turns["session-1"]) == 1 + + +@pytest.mark.asyncio +async def test_stream_deadline_closes_connection(monkeypatch, tmp_path): + upstream = GatedStream() + server, stream, requests = setup(monkeypatch, tmp_path, upstream) + monkeypatch.setattr("skillclaw.api_server._llm_request_timeout_seconds", lambda: 0.05) + parts = [part async for part in stream] + assert "504" in parts[-1] and upstream.closed + assert server._session_turns.get("session-1", []) == [] + assert len(requests) == 1 + + +@pytest.mark.asyncio +async def test_chat_route_uses_live_generator_and_disables_buffering(monkeypatch, tmp_path): + server = SkillClawAPIServer( + SkillClawConfig( + claw_type="none", + use_skills=False, + use_prm=False, + record_enabled=False, + record_dir=str(tmp_path), + ) + ) + seen = [] + + async def live(body, session_id, turn_type, session_done): + seen.append((body, session_id, turn_type, session_done)) + yield 'data: {"choices":[]}\n\n' + yield "data: [DONE]\n\n" + + monkeypatch.setattr(server, "_stream_chat_and_track", live) + async with httpx.AsyncClient(transport=httpx.ASGITransport(app=server.app), base_url="http://test") as client: + response = await client.post( + "/v1/chat/completions", + headers={"X-Session-Id": "explicit"}, + json={"messages": [{"role": "user", "content": "hi"}], "stream": True}, + ) + assert response.status_code == 200 + assert response.headers["x-accel-buffering"] == "no" + assert "[DONE]" in response.text and seen[0][1] == "explicit"