mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-26 13:16:24 +00:00
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 <willem.jiang@gmail.com>
This commit is contained in:
parent
eb7fc28010
commit
16a4f780c1
@ -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
|
||||
|
||||
@ -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
|
||||
|
||||
@ -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:
|
||||
|
||||
@ -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
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user