fix(middlewares): 标题模型推理不再泄漏到 UI 流 + open/ephemeral 会话避免无谓 DB 访问
- title: 标题模型用隔离 config(不继承父级 streaming callbacks),force_disable_thinking,<think> 块清洗,避免推理 token 进入实时流/运行日志 - uploads: 无文件的普通对话轮次直接跳过,不再解析线程归属(避免命中 DB) - thread_data: ephemeral 会话用当前登录用户解析数据目录 - thread_paths: open-/__skill-review__ 等合成线程视为 local-only,直接用当前用户 - frontend: 新增 build:test 构建脚本 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
1 parent
6aff165d09
commit
acc29c8121
5 files changed
+57
-28
No files matched your search
@@ -7,6 +7,7 @@
|
||||
"dev": "vite",
|
||||
"build": "node --max-old-space-size=8192 ./node_modules/vite/bin/vite.js build",
|
||||
"build:sit": "node --max-old-space-size=8192 ./node_modules/vite/bin/vite.js build --mode sit",
|
||||
"build:test": "node --max-old-space-size=8192 ./node_modules/vite/bin/vite.js build --mode test",
|
||||
"build:network": "node --max-old-space-size=8192 ./node_modules/vite/bin/vite.js build --mode network",
|
||||
"preview": "vite preview",
|
||||
"typecheck": "tsc --noEmit"
|
||||
|
||||
+10
-3
@@ -11,6 +11,7 @@ from langgraph.runtime import Runtime
|
||||
from deerflow.agents.thread_state import ThreadDataState
|
||||
from deerflow.config.paths import Paths, get_paths
|
||||
from deerflow.runtime.thread_paths import resolve_path_user_id
|
||||
from deerflow.runtime.user_context import get_effective_user_id
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@@ -81,10 +82,15 @@ class ThreadDataMiddleware(AgentMiddleware[ThreadDataMiddlewareState]):
|
||||
@override
|
||||
def before_agent(self, state: ThreadDataMiddlewareState, runtime: Runtime) -> dict | None:
|
||||
context = runtime.context or {}
|
||||
configurable = {}
|
||||
try:
|
||||
config = get_config()
|
||||
configurable = config.get("configurable", {}) or {}
|
||||
except RuntimeError:
|
||||
pass
|
||||
thread_id = context.get("thread_id")
|
||||
if thread_id is None:
|
||||
config = get_config()
|
||||
thread_id = config.get("configurable", {}).get("thread_id")
|
||||
thread_id = configurable.get("thread_id")
|
||||
|
||||
if thread_id is None:
|
||||
raise ValueError("Thread ID is required in runtime context or config.configurable")
|
||||
@@ -93,7 +99,8 @@ class ThreadDataMiddleware(AgentMiddleware[ThreadDataMiddlewareState]):
|
||||
# thread creator so every account that shares the conversation writes to
|
||||
# the same physical user-data directory. Personal threads resolve to the
|
||||
# current login unchanged.
|
||||
user_id = resolve_path_user_id(thread_id)
|
||||
is_ephemeral = bool(context.get("ephemeral") or configurable.get("ephemeral"))
|
||||
user_id = get_effective_user_id() if is_ephemeral else resolve_path_user_id(thread_id)
|
||||
|
||||
if self._lazy_init:
|
||||
# Lazy initialization: only compute paths, don't create directories
|
||||
|
||||
+31
-23
@@ -6,7 +6,6 @@ from typing import TYPE_CHECKING, Any, NotRequired, override
|
||||
|
||||
from langchain.agents import AgentState
|
||||
from langchain.agents.middleware import AgentMiddleware
|
||||
from langgraph.config import get_config
|
||||
from langgraph.runtime import Runtime
|
||||
|
||||
from deerflow.config.title_config import get_title_config
|
||||
@@ -139,13 +138,29 @@ class TitleMiddleware(AgentMiddleware[TitleMiddlewareState]):
|
||||
return text.strip()
|
||||
|
||||
def _parse_title(self, content: object) -> str:
|
||||
"""Normalize model output into a clean title string."""
|
||||
"""Normalize model output into a clean title string (strips <think> blocks)."""
|
||||
config = self._get_title_config()
|
||||
title_content = self._normalize_content(content)
|
||||
title_content = self._strip_think_tags(title_content)
|
||||
title_content = self._strip_think_tags(self._normalize_content(content))
|
||||
title = title_content.strip().strip('"').strip("'")
|
||||
return title[: config.max_chars] if len(title) > config.max_chars else title
|
||||
|
||||
def _isolated_config(self) -> dict[str, Any]:
|
||||
"""Build a RunnableConfig that does NOT inherit the parent's streaming callbacks.
|
||||
|
||||
The leak that pushed title-generation reasoning into the live UI stream came
|
||||
from inheriting the parent RunnableConfig (and with it the SSE/streaming
|
||||
callbacks) for the title model call. Here we deliberately start from an empty
|
||||
config with ``callbacks: []`` so the title model's tokens — including any
|
||||
``<think>`` reasoning from a reasoning-only deployment — never reach the
|
||||
frontend stream or the run journal as conversation output. The stored title
|
||||
is additionally cleaned by ``_parse_title``/``_strip_think_tags``.
|
||||
"""
|
||||
return {
|
||||
"run_name": "title_agent",
|
||||
"tags": ["middleware:title"],
|
||||
"callbacks": [],
|
||||
}
|
||||
|
||||
def _fallback_title(self, user_msg: str) -> str:
|
||||
config = self._get_title_config()
|
||||
fallback_chars = min(config.max_chars, 50)
|
||||
@@ -153,21 +168,6 @@ class TitleMiddleware(AgentMiddleware[TitleMiddlewareState]):
|
||||
return user_msg[:fallback_chars].rstrip() + "..."
|
||||
return user_msg if user_msg else "New Conversation"
|
||||
|
||||
def _get_runnable_config(self) -> dict[str, Any]:
|
||||
"""Inherit the parent RunnableConfig and add middleware tag.
|
||||
|
||||
This ensures RunJournal identifies LLM calls from this middleware
|
||||
as ``middleware:title`` instead of ``lead_agent``.
|
||||
"""
|
||||
try:
|
||||
parent = get_config()
|
||||
except Exception:
|
||||
parent = {}
|
||||
config = {**parent}
|
||||
config["run_name"] = "title_agent"
|
||||
config["tags"] = [*(config.get("tags") or []), "middleware:title"]
|
||||
return config
|
||||
|
||||
def _provisional_title_result(self, state: TitleMiddlewareState) -> dict | None:
|
||||
"""Set an instant placeholder title from the first user message (no LLM)."""
|
||||
if not self._should_set_provisional_title(state):
|
||||
@@ -221,7 +221,15 @@ class TitleMiddleware(AgentMiddleware[TitleMiddlewareState]):
|
||||
logger.debug("Could not override extra_body to disable thinking on title model", exc_info=True)
|
||||
|
||||
async def _agenerate_title_result(self, state: TitleMiddlewareState) -> dict | None:
|
||||
"""Generate a title asynchronously and fall back locally on failure."""
|
||||
"""Polish the title with an LLM, without leaking its reasoning to the UI.
|
||||
|
||||
The title model is invoked with an *isolated* config (``_isolated_config``)
|
||||
that does not inherit the parent's streaming callbacks, so even a
|
||||
reasoning-only deployment's ``<think>`` tokens never reach the live UI
|
||||
stream or the run journal. Thinking is also force-disabled on the model,
|
||||
and the final stored title is stripped of any residual ``<think>`` blocks.
|
||||
On any failure we fall back to the local question-derived title.
|
||||
"""
|
||||
if not self._should_generate_title(state):
|
||||
return None
|
||||
|
||||
@@ -229,7 +237,7 @@ class TitleMiddleware(AgentMiddleware[TitleMiddlewareState]):
|
||||
prompt, user_msg = self._build_title_prompt(state)
|
||||
|
||||
try:
|
||||
model_kwargs = {"thinking_enabled": False}
|
||||
model_kwargs: dict[str, Any] = {"thinking_enabled": False, "force_disable_thinking": True}
|
||||
if self._app_config is not None:
|
||||
model_kwargs["app_config"] = self._app_config
|
||||
if config.model_name:
|
||||
@@ -237,12 +245,12 @@ class TitleMiddleware(AgentMiddleware[TitleMiddlewareState]):
|
||||
else:
|
||||
model = create_chat_model(**model_kwargs)
|
||||
self._force_disable_thinking(model)
|
||||
response = await model.ainvoke(prompt, config=self._get_runnable_config())
|
||||
response = await model.ainvoke(prompt, config=self._isolated_config())
|
||||
title = self._parse_title(response.content)
|
||||
if title:
|
||||
return {"title": title, "title_provisional": False}
|
||||
except Exception:
|
||||
logger.debug("Failed to generate async title; falling back to local title", exc_info=True)
|
||||
logger.debug("Failed to generate async title via LLM; falling back to local title", exc_info=True)
|
||||
return {"title": self._fallback_title(user_msg), "title_provisional": False}
|
||||
|
||||
@override
|
||||
|
||||
+7
@@ -285,6 +285,13 @@ class UploadsMiddleware(AgentMiddleware[UploadsMiddlewareState]):
|
||||
if not isinstance(last_message, HumanMessage):
|
||||
return None
|
||||
|
||||
# Only turns with fresh upload metadata need the uploads directory.
|
||||
# Avoid resolving the thread owner (which may hit the DB) for ordinary
|
||||
# no-file chat turns, including ephemeral open-chat requests.
|
||||
kwargs_files = (last_message.additional_kwargs or {}).get("files")
|
||||
if not isinstance(kwargs_files, list) or not kwargs_files:
|
||||
return None
|
||||
|
||||
# Resolve uploads directory for existence checks
|
||||
thread_id = (runtime.context or {}).get("thread_id")
|
||||
if thread_id is None:
|
||||
|
||||
@@ -42,6 +42,12 @@ logger = logging.getLogger(__name__)
|
||||
_owner_cache: dict[str, str] = {}
|
||||
_owner_cache_lock = threading.Lock()
|
||||
_OWNER_CACHE_MAX = 4096
|
||||
_LOCAL_ONLY_THREAD_PREFIXES = ("open-", "__skill-review__")
|
||||
|
||||
|
||||
def _is_local_only_thread(thread_id: str | None) -> bool:
|
||||
"""Return True for synthetic threads that are never persisted in threads_meta."""
|
||||
return bool(thread_id) and str(thread_id).startswith(_LOCAL_ONLY_THREAD_PREFIXES)
|
||||
|
||||
|
||||
def _cache_get(thread_id: str) -> str | None:
|
||||
@@ -112,7 +118,7 @@ def resolve_path_user_id(thread_id: str | None) -> str:
|
||||
share the conversation read/write the same physical directory. For every
|
||||
other case it is the current login (``get_effective_user_id``).
|
||||
"""
|
||||
if not thread_id:
|
||||
if not thread_id or _is_local_only_thread(thread_id):
|
||||
return get_effective_user_id()
|
||||
cached = _cache_get(thread_id)
|
||||
if cached is not None:
|
||||
@@ -132,7 +138,7 @@ def resolve_path_user_id(thread_id: str | None) -> str:
|
||||
|
||||
async def aresolve_path_user_id(thread_id: str | None) -> str:
|
||||
"""Async variant of :func:`resolve_path_user_id` for FastAPI handlers."""
|
||||
if not thread_id:
|
||||
if not thread_id or _is_local_only_thread(thread_id):
|
||||
return get_effective_user_id()
|
||||
cached = _cache_get(thread_id)
|
||||
if cached is not None:
|
||||
|
||||
Reference in new issue
Block a user