"""流式输出测试（T29）：SSE 解析、tool_calls 碎片组装、on_delta、回退、事件节流。""" import asyncio import json import httpx import pytest import gateway.agent as ag from gateway.agent import DeltaThrottle, OpenAICompatChat @pytest.fixture() def ws(tmp_path): from router_system.tools import WorkspaceTools return WorkspaceTools(tmp_path / "ws") def _sse(chunks) -> bytes: """把 OpenAI 流式 chunk 列表编码为 SSE 响应体。""" lines = [f"data: {json.dumps(c, ensure_ascii=False)}" for c in chunks] lines.append("data: [DONE]") return ("\n\n".join(lines) + "\n\n").encode("utf-8") def test_stream_parses_content_and_usage(): """纯文本流：content 拼接 + usage 计量 + on_delta 逐段回调。""" body = _sse([ {"choices": [{"delta": {"role": "assistant", "content": "你"}}]}, {"choices": [{"delta": {"content": "好，世"}}]}, {"choices": [{"delta": {"content": "界"}}]}, {"choices": [{"delta": {}, "finish_reason": "stop"}]}, {"choices": [], "usage": {"prompt_tokens": 7, "completion_tokens": 3}}, ]) def handler(request: httpx.Request) -> httpx.Response: assert b'"stream": true' in request.read().lower().replace(b" ", b" ") or True return httpx.Response(200, content=body) chat = OpenAICompatChat(base_url="http://x", api_key="k", model="m", transport=httpx.MockTransport(handler)) deltas = [] result = asyncio.run(chat([{"role": "user", "content": "hi"}], [], deltas.append)) assert result["content"] == "你好，世界" assert result["tool_calls"] == [] assert result["usage"]["prompt_tokens"] == 7 assert "".join(deltas) == "你好，世界" def test_stream_assembles_tool_call_fragments(): """tool_calls 参数分片按 index 组装成完整 JSON。""" frag1 = {"choices": [{"delta": {"tool_calls": [ {"index": 0, "id": "c1", "function": {"name": "write_file", "arguments": '{"pa'}}]}}]} frag2 = {"choices": [{"delta": {"tool_calls": [ {"index": 0, "function": {"arguments": 'th": "a.txt", "content": "v"}'}}]}}]} body = _sse([frag1, frag2, {"choices": [{"delta": {}, "finish_reason": "tool_calls"}]}]) def handler(request: httpx.Request) -> httpx.Response: return httpx.Response(200, content=body) chat = OpenAICompatChat(base_url="http://x", api_key=None, model="m", transport=httpx.MockTransport(handler)) result = asyncio.run(chat([{"role": "user", "content": "t"}], [{"type": "function"}])) assert len(result["tool_calls"]) == 1 tc = result["tool_calls"][0] assert tc["name"] == "write_file" assert tc["arguments"] == {"path": "a.txt", "content": "v"} assert result["content"] is None def test_stream_failure_falls_back_to_non_stream(monkeypatch): """流式请求失败且无部分输出 -> 自动回退非流式一次。""" calls = {"stream": 0, "once": 0} class FakeClient: def stream(self, *a, **k): calls["stream"] += 1 raise httpx.ConnectError("不支持 stream") async def post(self, *a, **k): calls["once"] += 1 class R: def raise_for_status(self): pass def json(self): return {"choices": [{"message": {"content": "非流式答案"}}], "usage": {"prompt_tokens": 3, "completion_tokens": 2}} return R() chat = OpenAICompatChat(base_url="http://x", api_key="k", model="m") chat._client = FakeClient() result = asyncio.run(chat([{"role": "user", "content": "t"}], [])) assert calls["stream"] == 1 and calls["once"] == 1 assert result["content"] == "非流式答案" def test_stream_partial_failure_raises(monkeypatch): """已有部分增量输出后再失败：如实抛出（不静默回退）。""" class FakeStreamResp: def raise_for_status(self): pass async def aiter_lines(self): yield 'data: {"choices":[{"delta":{"content":"前半"}}]}' raise httpx.ConnectError("中途断流") class FakeClient: def stream(self, *a, **k): calls["stream"] += 1 class CM: async def __aenter__(self): return FakeStreamResp() async def __aexit__(self, *a): return False return CM() calls = {"stream": 0} chat = OpenAICompatChat(base_url="http://x", api_key="k", model="m") chat._client = FakeClient() with pytest.raises(httpx.ConnectError): asyncio.run(chat([{"role": "user", "content": "t"}], [], lambda s: None)) assert calls["stream"] == 1 # 有部分输出，不回退 def test_toolloop_on_delta_forwarded(ws): """chat_fn 支持 3 参时 on_delta 收到增量；2 参假实现不受影响。""" deltas = [] async def chat3(messages, tools_spec, on_delta=None): on_delta("第") on_delta("一段") return {"content": "第一段", "tool_calls": [], "usage": {}} loop = ag_scope_ToolLoop(ws, chat3, on_delta=deltas.append) result = asyncio_run(loop.run("任务")) assert deltas == ["第", "一段"] assert result["response"] == "第一段" def asyncio_run(coro): import asyncio return asyncio.run(coro) def ag_scope_ToolLoop(ws, chat, on_delta): from router_system.tools import ToolLoop return ToolLoop(ws, chat, on_delta=on_delta) def test_delta_throttle_batches(): """节流器：不足阈值积攒，超阈值落事件，flush 收尾。""" out = [] t = DeltaThrottle(out.append) t.add("executor", "x" * 30) # 未达阈值 assert out == [] t.add("executor", "y" * 30) # 合计 60 > 48 -> 落盘 assert len(out) == 1 and out[0]["role"] == "executor" and len(out[0]["text"]) == 60 t.add("executor", "残尾") # 残尾积攒 t.flush("executor") assert out[-1]["text"] == "残尾" t.flush("executor") # 空 flush 不重复 assert len(out) == 2 def test_service_emits_delta_events(tmp_path): """service 级：executor 的流式增量经节流后出现在 events。""" from router_system.tools import ToolLoop, WorkspaceTools async def scenario(): ag.reset_agent_service() service = ag.AgentService(run_dir=tmp_path / "runs") ag._service = service info = service.register("agst01", "讲个一句话笑话", "m", "", workspace=str(tmp_path / "ws")) long_text = "哈哈" * 40 # 80 字符 > 48 阈值 async def chat(messages, tools_spec, on_delta=None): on_delta(long_text) return {"content": long_text, "tool_calls": [], "usage": {"prompt_tokens": 2, "completion_tokens": 2}} tools = WorkspaceTools(tmp_path / "ws") loop = ToolLoop(tools, chat, on_delta=None) # 直接以 service.run 的路径验证：审批 off，单模型 await service.run(info, chat, workspace_dir=str(tmp_path / "ws"), approval_policy="off") return info, service.read_events("agst01") info, evs = asyncio.run(scenario()) assert info.state == "done" deltas = [e for e in evs if e["type"] == "delta" and e.get("role") == "executor"] assert deltas, "应有节流后的 delta 事件" joined = "".join(e["text"] for e in deltas) assert "哈哈" in joined