tzt
ddef8bec1c
- OpenAICompatChat 默认 stream=True：httpx 流式解析 OpenAI chunk， tool_calls 碎片按 index 组装（不在正文展示），include_usage 计量； 失败自动回退非流式一次（已有部分增量输出则如实抛出）； 补 raise_for_status（修复流式 4xx 误入回退的缺陷） - ToolLoop 增 on_delta（签名探测兼容 2/3 参 chat_fn）；_loads_json_object 宽松解析 - DeltaThrottle >=48 字符节流落 delta 事件；单模型与两级模式（含规划者）接线 - 前端：streamText 打字机渲染 + 光标动画；工具/阶段事件到达时清空归档 - 配套修复：AgentView 闭包持有 push 前原始对象导致响应式丢失、过程事件不渲染 - 测试 +7（SSE 解析/碎片组装/回退/部分失败抛出/on_delta/节流/service delta 事件）， 全量 296 passed
197 lines
7.5 KiB
Python
197 lines
7.5 KiB
Python
"""流式输出测试（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
|