mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-31 07:30:43 +00:00
fix(gateway): read the live config off the loop for thread token usage (#7727)
* fix(gateway): read the live config off the loop for thread token usage
build_context_usage resolves the model capacity through deps.get_config,
which revalidates config.yaml and extensions_config.json on every call, so
each /threads/{id}/token-usage request paid that file IO on the shared
Gateway loop.
* docs(changelog): record the thread token-usage config read move
---------
Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
This commit is contained in:
parent
cdc3e92d6b
commit
5d2749b05b
@ -26,6 +26,12 @@ This section accumulates work toward the **2.3.0** milestone
|
||||
plaintext, which with authorization disabled (the default) is every
|
||||
signed-in user. Non-secret knobs, `api_key_env` names and unset values are
|
||||
unchanged. ([#7665])
|
||||
- **gateway:** `GET /threads/{thread_id}/token-usage` now resolves the live
|
||||
config in a worker thread. `build_context_usage` called the accessor inline,
|
||||
and `get_app_config()` revalidates on every call — it digests the whole
|
||||
`config.yaml`, then `extensions_config.json` — so a config edit landing just
|
||||
before a thread view stalled the Gateway event loop that runs, streams and
|
||||
every other thread share. The response is unchanged. ([#7727])
|
||||
|
||||
## [2.2.0] — 2026-10-29
|
||||
|
||||
@ -15934,3 +15940,4 @@ with **180 merged pull requests** since the first 2.0 milestone tag.
|
||||
[#7659]: https://github.com/bytedance/deer-flow/pull/7659
|
||||
[#7665]: https://github.com/bytedance/deer-flow/pull/7665
|
||||
[#7723]: https://github.com/bytedance/deer-flow/pull/7723
|
||||
[#7727]: https://github.com/bytedance/deer-flow/pull/7727
|
||||
|
||||
@ -24,6 +24,10 @@
|
||||
此前它们按加载结果原样返回配置,而 `$VAR` 引用此时已被解析,因此 DeerMem 提取模型的密钥
|
||||
或 Honcho 的 `api_key` 会以明文返回给所有拥有 `memory:read` 的调用方;在默认关闭授权时,
|
||||
这就是所有已登录用户。非机密配置项、`api_key_env` 变量名和未设置的值保持不变。([#7665])
|
||||
- **网关:** `GET /threads/{thread_id}/token-usage` 现在在工作线程里解析实时配置。
|
||||
此前 `build_context_usage` 直接内联调用该访问器,而 `get_app_config()` 每次调用都会重新校验
|
||||
——它先对整个 `config.yaml` 取摘要,再对 `extensions_config.json` 做同样的事——所以在有人打开
|
||||
会话视图前刚好改了配置,就会让运行、消息流与其他线程共用的网关事件循环整段停顿。响应内容不变。([#7727])
|
||||
|
||||
## [2.2.0] — 2026-10-29
|
||||
|
||||
@ -12701,3 +12705,4 @@ DeerFlow 2.0 是围绕"超级智能体"框架的彻底重写,核心包含子
|
||||
[#7659]: https://github.com/bytedance/deer-flow/pull/7659
|
||||
[#7665]: https://github.com/bytedance/deer-flow/pull/7665
|
||||
[#7723]: https://github.com/bytedance/deer-flow/pull/7723
|
||||
[#7727]: https://github.com/bytedance/deer-flow/pull/7727
|
||||
|
||||
@ -66,7 +66,10 @@ def build_context_usage_payload(*, token_count: int, max_context_tokens: int | N
|
||||
async def build_context_usage(request: Request, thread_id: str, run_store: Any, user_id: str | None = None) -> dict[str, Any] | None:
|
||||
"""Return approximate usage for the latest materialized thread checkpoint."""
|
||||
try:
|
||||
app_config = get_config()
|
||||
# ``get_config`` revalidates ``config.yaml`` and
|
||||
# ``extensions_config.json`` on every call (stat + full-file hash), so
|
||||
# reading it here would put that IO on the loop every thread view pays.
|
||||
app_config = await asyncio.to_thread(get_config)
|
||||
except HTTPException:
|
||||
return None
|
||||
|
||||
|
||||
132
backend/tests/blocking_io/test_context_usage_config_read.py
Normal file
132
backend/tests/blocking_io/test_context_usage_config_read.py
Normal file
@ -0,0 +1,132 @@
|
||||
"""Regression anchor: ``/threads/{id}/token-usage`` must read the live config off the loop.
|
||||
|
||||
``build_context_usage`` resolves the model capacity for its payload through
|
||||
``app.gateway.deps.get_config``, which routes to ``get_app_config()``. That
|
||||
loader revalidates on every call — it stats and hashes the whole
|
||||
``config.yaml``, then does the same for ``extensions_config.json`` (see
|
||||
**Config Caching** in ``deerflow/config/AGENTS.md``) — and reparses both after
|
||||
an edit. ``GET /threads/{thread_id}/token-usage`` builds this payload on every
|
||||
request, while runs, streams and other threads share the same Gateway loop, so
|
||||
a view requested just after a config edit made that request stall the loop for
|
||||
the whole reload. The token count a few lines below the read was already
|
||||
offloaded with ``asyncio.to_thread``; the config read now joins it.
|
||||
|
||||
Each anchor performs a real filesystem read and asserts the capacity that only
|
||||
the bytes on disk can produce, so it cannot pass because the read was stubbed
|
||||
away, and it records the calling thread so a dropped wrapper is attributed.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import threading
|
||||
from collections.abc import AsyncIterator
|
||||
from pathlib import Path
|
||||
from types import SimpleNamespace
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
import pytest_asyncio
|
||||
from blockbuster import BlockingError
|
||||
from fastapi import HTTPException
|
||||
from langchain_core.messages import HumanMessage
|
||||
|
||||
from app.gateway import context_usage
|
||||
|
||||
pytestmark = pytest.mark.asyncio
|
||||
|
||||
_CAPACITY_ON_DISK = 4096
|
||||
_CONFIG_TEXT = f"model: disk-model\ntoken_capacity: {_CAPACITY_ON_DISK}\n"
|
||||
|
||||
|
||||
class _RecordingConfigLoader:
|
||||
"""``get_config`` stand-in that reads from disk, and remembers where it ran."""
|
||||
|
||||
def __init__(self, path: Path) -> None:
|
||||
self.path = path
|
||||
self.calls = 0
|
||||
self.on_loop_thread = False
|
||||
|
||||
def __call__(self) -> SimpleNamespace:
|
||||
self.calls += 1
|
||||
self.on_loop_thread = threading.current_thread() is threading.main_thread()
|
||||
name, capacity = self.path.read_text(encoding="utf-8").splitlines()
|
||||
model_name = name.split(":", 1)[1].strip()
|
||||
return SimpleNamespace(
|
||||
models=[SimpleNamespace(name=model_name)],
|
||||
get_model_config=lambda requested: SimpleNamespace(context_window=int(capacity.split(":", 1)[1])) if requested == model_name else None,
|
||||
)
|
||||
|
||||
|
||||
class _Accessor:
|
||||
async def aget(self, config: dict[str, Any]) -> SimpleNamespace:
|
||||
return SimpleNamespace(values={"messages": [HumanMessage(content="context window probe")]})
|
||||
|
||||
|
||||
async def _stub_accessor_builder(request, *, thread_id: str):
|
||||
return _Accessor(), {"configurable": {"thread_id": thread_id}}
|
||||
|
||||
|
||||
class _RunStore:
|
||||
async def list_by_thread(self, thread_id: str, *, limit: int = 1, user_id: str | None = None):
|
||||
return [{"model_name": "disk-model"}]
|
||||
|
||||
|
||||
def _write_capacity_file(path: Path) -> Path:
|
||||
path.write_text(_CONFIG_TEXT, encoding="utf-8")
|
||||
return path
|
||||
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def loader(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> AsyncIterator[_RecordingConfigLoader]:
|
||||
# The strict gate wraps setup as well as the call, so the fixture's own
|
||||
# write goes through a worker.
|
||||
config_file = await asyncio.to_thread(_write_capacity_file, tmp_path / "capacity.yaml")
|
||||
recorder = _RecordingConfigLoader(config_file)
|
||||
monkeypatch.setattr(context_usage, "get_config", recorder)
|
||||
monkeypatch.setattr(context_usage, "build_thread_checkpoint_state_accessor", _stub_accessor_builder)
|
||||
yield recorder
|
||||
await asyncio.to_thread(config_file.unlink)
|
||||
|
||||
|
||||
async def test_build_context_usage_reads_capacity_off_the_loop(loader: _RecordingConfigLoader) -> None:
|
||||
payload = await context_usage.build_context_usage(SimpleNamespace(), "thread-1", _RunStore())
|
||||
|
||||
assert payload is not None
|
||||
# The capacity only the file on disk can produce: the read was not stubbed away.
|
||||
assert payload["max_context_tokens"] == _CAPACITY_ON_DISK
|
||||
assert payload["token_count"] > 0
|
||||
assert payload["percentage"] == pytest.approx(round(payload["token_count"] / _CAPACITY_ON_DISK * 100, 1))
|
||||
assert loader.calls == 1
|
||||
assert not loader.on_loop_thread, "the live config read ran on the event loop thread"
|
||||
|
||||
|
||||
async def test_offloaded_read_still_reports_an_unavailable_config(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""Control: moving the read to a worker must not turn a 503 into a payload."""
|
||||
|
||||
def raise_unavailable() -> SimpleNamespace:
|
||||
raise HTTPException(status_code=503, detail="Configuration not available")
|
||||
|
||||
monkeypatch.setattr(context_usage, "get_config", raise_unavailable)
|
||||
monkeypatch.setattr(context_usage, "build_thread_checkpoint_state_accessor", _stub_accessor_builder)
|
||||
|
||||
assert await context_usage.build_context_usage(SimpleNamespace(), "thread-1", _RunStore()) is None
|
||||
|
||||
|
||||
async def test_inline_config_read_actually_trips_the_gate(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""Meta-check: the gate is live around this module, so the anchor above is not vacuous.
|
||||
|
||||
The strict gate only trips when the blocking read sits under an ``app`` or
|
||||
``deerflow`` frame, so this calls the loader ``deps.get_config`` routes to.
|
||||
That accessor maps every failure to 503, which would hide the
|
||||
``BlockingError``, hence the loader itself.
|
||||
"""
|
||||
from deerflow.config.app_config import get_app_config, reset_app_config
|
||||
|
||||
monkeypatch.setenv("DEER_FLOW_CONFIG_PATH", str(tmp_path / "capacity.yaml"))
|
||||
await asyncio.to_thread(reset_app_config)
|
||||
try:
|
||||
with pytest.raises(BlockingError):
|
||||
get_app_config()
|
||||
finally:
|
||||
await asyncio.to_thread(reset_app_config)
|
||||
Loading…
x
Reference in New Issue
Block a user