From 16a4f780c13a412e7cbcb0004198a2dd31e6165d Mon Sep 17 00:00:00 2001 From: Hyeonsang Cho Date: Mon, 12 Oct 2026 18:38:24 +0900 Subject: [PATCH] fix(gateway): read subagent batch exports in small pages (#6706) * fix(gateway): read subagent batch exports in small pages The results.jsonl export loaded 500 full rows per page before streaming them. A row can hold max_result_chars of report plus as much evidence JSON (result_artifact), so with the default limit one request peaked around 280 MB and decoding a page's JSON columns stalled the event loop for about half a second; at the 1,000,000 ceiling a page was roughly 1.7 GB. Read 10 rows per page instead. Output and ordering are unchanged; a full export of 1,000 default-sized rows now peaks at 6 MB with a 46 ms worst stall and takes the same time. * docs(changelog): reference #6706 --------- Co-authored-by: Willem Jiang --- CHANGELOG.md | 9 ++++++++ CHANGELOG_zh.md | 7 ++++++ .../app/gateway/routers/subagent_batches.py | 6 ++++- backend/tests/test_subagent_batches_router.py | 23 +++++++++++++++++++ 4 files changed, 44 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 8667fd7198..2717c6608c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5807,6 +5807,14 @@ This release closes that milestone with **621 merged pull requests**. the trailing `-07` as a UTC offset and displayed UTC midnight instead: 08:00 in `Asia/Shanghai`, and the previous evening west of UTC, in the task list, detail, chat card and the edit form's prefilled time. ([#6692]) +- **gateway:** The subagent batch JSONL export + (`GET /api/threads/{thread_id}/subagent-batches/{batch_id}/results.jsonl`) + now reads stored results 10 rows at a time instead of 500. Each row can hold + `max_result_chars` of report plus as much evidence JSON, and the export loads + a whole page before streaming it, so one request held about 280 MB with the + default limit (roughly 1.7 GB at the 1,000,000 ceiling) and stalled the event + loop for over half a second per page. The exported lines and their order are + unchanged. ([#6706]) ### Security @@ -11582,3 +11590,4 @@ with **180 merged pull requests** since the first 2.0 milestone tag. [#6691]: https://github.com/bytedance/deer-flow/pull/6691 [#6692]: https://github.com/bytedance/deer-flow/pull/6692 [#6702]: https://github.com/bytedance/deer-flow/pull/6702 +[#6706]: https://github.com/bytedance/deer-flow/pull/6706 diff --git a/CHANGELOG_zh.md b/CHANGELOG_zh.md index fc93b713cd..d99d74971c 100644 --- a/CHANGELOG_zh.md +++ b/CHANGELOG_zh.md @@ -4441,6 +4441,12 @@ 任务，现在显示为任务时区当天的零点，与服务端实际执行时间一致。此前界面把末尾的 `-07` 误认为 UTC 偏移量，显示成 UTC 零点：在 `Asia/Shanghai` 显示为 08:00，在 UTC 以西的时区则显示为前一天晚上；任务列表、详情、对话卡片和编辑表单的预填时间都受影响。([#6692]) +- **网关：** 子代理批次 JSONL 导出 + （`GET /api/threads/{thread_id}/subagent-batches/{batch_id}/results.jsonl`） + 现在每次读取 10 行已存储结果，而不是 500 行。每行最多可包含 `max_result_chars` + 的报告和同等长度的证据 JSON，而导出会先把整页载入内存再开始输出，因此默认上限下 + 单个请求约占用 280 MB（上限 1,000,000 时约 1.7 GB），并且每页会阻塞事件循环 + 半秒以上。导出的行及其顺序不变。([#6706]) ### 安全 @@ -9277,3 +9283,4 @@ DeerFlow 2.0 是围绕"超级智能体"框架的彻底重写，核心包含子 [#6691]: https://github.com/bytedance/deer-flow/pull/6691 [#6692]: https://github.com/bytedance/deer-flow/pull/6692 [#6702]: https://github.com/bytedance/deer-flow/pull/6702 +[#6706]: https://github.com/bytedance/deer-flow/pull/6706 diff --git a/backend/app/gateway/routers/subagent_batches.py b/backend/app/gateway/routers/subagent_batches.py index 705aff3b21..4cb575f557 100644 --- a/backend/app/gateway/routers/subagent_batches.py +++ b/backend/app/gateway/routers/subagent_batches.py @@ -18,6 +18,10 @@ from deerflow.utils.thread_id import ThreadId router = APIRouter(prefix="/api/threads/{thread_id}/subagent-batches", tags=["subagent-batches"]) _ITEM_STATUSES = {"pending", "queued", "leased", "running", "succeeded", "failed", "cancelled"} +# A full row holds up to max_result_chars of report plus as much evidence JSON: +# 2M characters at the config ceiling. Export loads one page into memory (and +# decodes its JSON on the event loop) before streaming it, so keep pages small. +_RESULTS_EXPORT_PAGE_SIZE = 10 async def _user_id(request: Request) -> str: @@ -114,7 +118,7 @@ async def export_batch_results(thread_id: ThreadId, batch_id: str, request: Requ batch_id, user_id=user_id, offset=offset, - limit=500, + limit=_RESULTS_EXPORT_PAGE_SIZE, include_result=True, ) if not page: diff --git a/backend/tests/test_subagent_batches_router.py b/backend/tests/test_subagent_batches_router.py index b15a02ad2d..08c9b6a2b9 100644 --- a/backend/tests/test_subagent_batches_router.py +++ b/backend/tests/test_subagent_batches_router.py @@ -29,6 +29,7 @@ class Repository: {"id": "item-2", "batch_id": "batch-1", "item_key": "two", "status": "failed", "error": "bad"}, ] self.include_result_calls = [] + self.limits = [] async def get_batch(self, batch_id, *, user_id): if batch_id != self.batch["id"] or user_id != self.batch["user_id"]: @@ -42,6 +43,7 @@ class Repository: async def list_items(self, batch_id, *, user_id, offset=0, limit=100, status=None, include_result=False): assert batch_id == "batch-1" and user_id == "user-1" self.include_result_calls.append(include_result) + self.limits.append(limit) values = self.items if status is not None: values = [item for item in values if item["status"] == status] @@ -124,3 +126,24 @@ async def test_jsonl_export_streams_item_results(monkeypatch) -> None: assert [row["id"] for row in rows] == ["item-1", "item-2"] assert rows[0]["result"] == "done" assert repo.include_result_calls and all(repo.include_result_calls) + + +@pytest.mark.asyncio +async def test_jsonl_export_reads_full_results_in_small_pages(monkeypatch) -> None: + # A full row can hold max_result_chars of report plus as much evidence + # JSON, up to 2M characters at the config ceiling; 500-row pages + # materialized hundreds of MB per request before the first line streamed. + repo = Repository() + repo.items = [{"id": f"item-{index}", "batch_id": "batch-1", "status": "succeeded", "result": str(index)} for index in range(25)] + monkeypatch.setattr(subagent_batches, "get_current_user", AsyncMock(return_value="user-1")) + + response = await subagent_batches.export_batch_results.__wrapped__( + thread_id="thread-1", + batch_id="batch-1", + request=_request(repo), + ) + payload = b"".join([chunk async for chunk in response.body_iterator]).decode() + + assert [json.loads(line)["id"] for line in payload.splitlines()] == [f"item-{index}" for index in range(25)] + assert max(repo.limits) <= 10 + assert len(repo.limits) > 2