2124 lines
89 KiB
Python
2124 lines
89 KiB
Python
"""Run lifecycle service layer.
|
||
|
||
Centralizes the business logic for creating runs, formatting SSE
|
||
frames, and consuming stream bridge events. Router modules
|
||
(``thread_runs``, ``runs``) are thin HTTP handlers that delegate here.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import json
|
||
import logging
|
||
import os
|
||
import re
|
||
import time
|
||
import uuid
|
||
from collections.abc import Awaitable, Callable, Mapping
|
||
from typing import Any
|
||
|
||
from fastapi import HTTPException, Request
|
||
from langchain_core.messages import HumanMessage
|
||
|
||
from app.gateway.deps import get_concurrency_gate, get_run_context, get_run_manager, get_stream_bridge
|
||
from app.gateway.llmwiki_rag import (
|
||
LlmWikiRagContext,
|
||
attach_llmwiki_rag_to_graph_input,
|
||
build_llmwiki_rag_context,
|
||
build_llmwiki_smart_reasoning_context,
|
||
llmwiki_selection_declared,
|
||
llmwiki_selection_requested,
|
||
)
|
||
from app.gateway.utils import sanitize_log_param
|
||
from app.gateway.worker_pool.admission import bounded_worker_run
|
||
from deerflow.agents.middlewares.prompt_prefix_middleware import (
|
||
SMART_KNOWLEDGE_SCOPE_PROMPT_KEY,
|
||
)
|
||
from deerflow.config import get_app_config
|
||
from deerflow.config.agents_config import load_agent_config, validate_agent_id
|
||
from deerflow.config.system_settings import load_system_settings
|
||
from deerflow.runtime import (
|
||
END_SENTINEL,
|
||
HEARTBEAT_SENTINEL,
|
||
ConflictError,
|
||
DisconnectMode,
|
||
ModelConcurrencyLimitError,
|
||
RunManager,
|
||
RunRecord,
|
||
RunStatus,
|
||
StreamBridge,
|
||
UnsupportedStrategyError,
|
||
run_agent,
|
||
)
|
||
from deerflow.runtime.user_context import get_effective_user_id
|
||
from deerflow.tools.builtins.llmwiki_reasoning_tools import (
|
||
CLASSIC_LLMWIKI_TOOL_NAMES,
|
||
)
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
RUN_SCOPE_EXTERNAL_ROOT = "external_root"
|
||
RUN_SCOPE_ORCHESTRATION_CHILD = "orchestration_child"
|
||
RUN_SCOPE_BACKGROUND_CHILD = "background_child"
|
||
_INTERNAL_RUN_SCOPES = {RUN_SCOPE_ORCHESTRATION_CHILD, RUN_SCOPE_BACKGROUND_CHILD}
|
||
_ALL_RUN_SCOPES = {RUN_SCOPE_EXTERNAL_ROOT, *_INTERNAL_RUN_SCOPES}
|
||
_INTERNAL_RUN_SCOPE_HEADER = "x-deerflow-internal-run-scope-token"
|
||
_INTERNAL_RUN_SCOPE_TOKEN = os.getenv("DEERFLOW_INTERNAL_RUN_SCOPE_TOKEN", "").strip() or uuid.uuid4().hex
|
||
|
||
|
||
def internal_run_scope_headers() -> dict[str, str]:
|
||
"""Headers used by same-process loopback calls to mark internal child runs."""
|
||
|
||
return {_INTERNAL_RUN_SCOPE_HEADER: _INTERNAL_RUN_SCOPE_TOKEN}
|
||
|
||
|
||
def _metadata_run_scope(body: Any) -> str | None:
|
||
metadata = getattr(body, "metadata", None) or {}
|
||
if not isinstance(metadata, dict):
|
||
return None
|
||
raw = str(metadata.get("run_scope") or "").strip()
|
||
return raw or None
|
||
|
||
|
||
def _request_is_loopback(request: Request) -> bool:
|
||
client = getattr(request, "client", None)
|
||
host = str(getattr(client, "host", "") or "").strip()
|
||
return host in {"127.0.0.1", "::1", "localhost"}
|
||
|
||
|
||
def _has_internal_run_scope_header(request: Request) -> bool:
|
||
headers = getattr(request, "headers", None)
|
||
token = headers.get(_INTERNAL_RUN_SCOPE_HEADER) if headers is not None else None
|
||
return str(token or "") == _INTERNAL_RUN_SCOPE_TOKEN
|
||
|
||
|
||
def _resolve_run_scope(body: Any, request: Request, run_scope: str | None) -> str:
|
||
"""Return the trusted run scope used for concurrency accounting."""
|
||
|
||
if run_scope:
|
||
scope = str(run_scope).strip()
|
||
return scope if scope in _ALL_RUN_SCOPES else RUN_SCOPE_EXTERNAL_ROOT
|
||
|
||
scope = _metadata_run_scope(body)
|
||
if scope in _INTERNAL_RUN_SCOPES and _request_is_loopback(request) and _has_internal_run_scope_header(request):
|
||
return scope
|
||
if scope == RUN_SCOPE_EXTERNAL_ROOT:
|
||
return RUN_SCOPE_EXTERNAL_ROOT
|
||
return RUN_SCOPE_EXTERNAL_ROOT
|
||
|
||
|
||
async def _run_agent_with_concurrency_lease(
|
||
*,
|
||
concurrency_gate: Any | None,
|
||
lease_acquired: bool,
|
||
lease_metadata: dict[str, Any],
|
||
heartbeat_interval_seconds: int,
|
||
run_agent_kwargs: dict[str, Any],
|
||
after_success: Callable[[], Awaitable[None]] | None = None,
|
||
) -> None:
|
||
"""Run the agent while refreshing and finally releasing an optional lease."""
|
||
|
||
heartbeat_task: asyncio.Task | None = None
|
||
|
||
async def _heartbeat_loop() -> None:
|
||
if concurrency_gate is None or not lease_acquired:
|
||
return
|
||
interval = max(5, int(heartbeat_interval_seconds or 20))
|
||
while True:
|
||
await asyncio.sleep(interval)
|
||
await concurrency_gate.heartbeat(**lease_metadata)
|
||
|
||
if concurrency_gate is not None and lease_acquired:
|
||
heartbeat_task = asyncio.create_task(
|
||
_heartbeat_loop(),
|
||
name=f"concurrency-heartbeat:{lease_metadata.get('run_id')}",
|
||
)
|
||
|
||
try:
|
||
from app.gateway.worker_pool.admission import run_with_worker_admission
|
||
|
||
await run_with_worker_admission(run_agent, **run_agent_kwargs)
|
||
record = run_agent_kwargs.get("record")
|
||
if after_success is not None and getattr(record, "status", None) == RunStatus.success:
|
||
try:
|
||
await after_success()
|
||
except Exception: # noqa: BLE001 - memory extraction must never fail chat
|
||
logger.exception(
|
||
"Post-run success callback failed for run %s (non-fatal)",
|
||
lease_metadata.get("run_id"),
|
||
)
|
||
finally:
|
||
if heartbeat_task is not None:
|
||
heartbeat_task.cancel()
|
||
try:
|
||
await heartbeat_task
|
||
except asyncio.CancelledError:
|
||
pass
|
||
if concurrency_gate is not None and lease_acquired:
|
||
await concurrency_gate.release(**lease_metadata)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# SSE formatting
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
async def immediate_end_sse():
|
||
"""Yield a single terminal ``end`` SSE frame and stop.
|
||
|
||
Used when a run exists in the persistent store but not in the in-memory
|
||
registry — typically after a Gateway restart, or once a finished run's
|
||
record has been cleaned up. There is no live stream to join, so instead of
|
||
returning 404 (which the SDK's ``joinStream`` surfaces as a hard error and
|
||
never clears its reconnect key for) we stream a clean end. The SDK treats
|
||
this as success, clears ``lg:stream:{threadId}``, and the conversation
|
||
falls back to history rendering — killing the repeating switch-back 404.
|
||
"""
|
||
yield format_sse("end", None)
|
||
|
||
|
||
def format_sse(event: str, data: Any, *, event_id: str | None = None) -> str:
|
||
"""Format a single SSE frame.
|
||
|
||
Field order: ``event:`` -> ``data:`` -> ``id:`` (optional) -> blank line.
|
||
This matches the LangGraph Platform wire format consumed by the
|
||
``useStream`` React hook and the Python ``langgraph-sdk`` SSE decoder.
|
||
"""
|
||
payload = json.dumps(data, default=str, ensure_ascii=False)
|
||
parts = [f"event: {event}", f"data: {payload}"]
|
||
if event_id:
|
||
parts.append(f"id: {event_id}")
|
||
parts.append("")
|
||
parts.append("")
|
||
return "\n".join(parts)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Input / config helpers
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def normalize_stream_modes(raw: list[str] | str | None) -> list[str]:
|
||
"""Normalize the stream_mode parameter to a list.
|
||
|
||
Default matches what ``useStream`` expects: values + messages-tuple.
|
||
"""
|
||
if raw is None:
|
||
return ["values"]
|
||
if isinstance(raw, str):
|
||
return [raw]
|
||
return raw if raw else ["values"]
|
||
|
||
|
||
def _with_knowledge_retrieval_progress_stream(
|
||
stream_modes: list[str],
|
||
knowledge_retrieval_mode: str,
|
||
) -> list[str]:
|
||
"""Ensure smart retrieval custom events reach every gateway SSE client."""
|
||
|
||
normalized = list(dict.fromkeys(stream_modes))
|
||
if knowledge_retrieval_mode == "smart_reasoning" and "custom" not in normalized:
|
||
normalized.append("custom")
|
||
return normalized
|
||
|
||
|
||
def normalize_input(raw_input: dict[str, Any] | None) -> dict[str, Any]:
|
||
"""Convert LangGraph Platform input format to LangChain state dict."""
|
||
if raw_input is None:
|
||
return {}
|
||
messages = raw_input.get("messages")
|
||
if messages and isinstance(messages, list):
|
||
converted = []
|
||
for msg in messages:
|
||
if isinstance(msg, dict):
|
||
content = msg.get("content", "")
|
||
# Preserve additional_kwargs (e.g. uploaded `files`, `client_ts`,
|
||
# `prompt_prefix`) and optional id/name so they round-trip through
|
||
# the checkpoint to the UI and to UploadsMiddleware. Dropping them
|
||
# here previously made uploaded files invisible in the chat and
|
||
# broke the per-message ask-time timestamp.
|
||
msg_kwargs: dict[str, Any] = {"content": content}
|
||
extra = msg.get("additional_kwargs")
|
||
if isinstance(extra, dict) and extra:
|
||
# Retrieval evidence and traces are server-authored. If
|
||
# accepted from a client, a user could forge DevTools/
|
||
# reference-batches diagnostics or inject fake citations.
|
||
server_retrieval_keys = {
|
||
"knowledge_retrieval_trace",
|
||
"llmwiki_reference_sources",
|
||
"llmwiki_reference_query",
|
||
"llmwiki_knowledge_base_ids",
|
||
"assistant_knowledge_base_ids",
|
||
"llmwiki_knowledge_base_context",
|
||
SMART_KNOWLEDGE_SCOPE_PROMPT_KEY,
|
||
}
|
||
safe_extra = {key: value for key, value in extra.items() if key not in server_retrieval_keys}
|
||
if safe_extra:
|
||
msg_kwargs["additional_kwargs"] = safe_extra
|
||
if msg.get("id"):
|
||
msg_kwargs["id"] = msg["id"]
|
||
if msg.get("name"):
|
||
msg_kwargs["name"] = msg["name"]
|
||
# TODO: handle other message types (system, ai, tool); for now all
|
||
# inbound dict messages are treated as human turns.
|
||
converted.append(HumanMessage(**msg_kwargs))
|
||
else:
|
||
converted.append(msg)
|
||
return {**raw_input, "messages": converted}
|
||
return raw_input
|
||
|
||
|
||
def _plain_fixed_question(graph_input: dict[str, Any]) -> str | None:
|
||
"""Return the latest plain-text human turn eligible for exact matching.
|
||
|
||
Messages carrying files or multimodal blocks are intentionally excluded:
|
||
their answer depends on more than the visible question text, so replaying a
|
||
global fixed answer would be surprising and unsafe.
|
||
"""
|
||
|
||
messages = graph_input.get("messages")
|
||
if not isinstance(messages, list) or not messages:
|
||
return None
|
||
message = messages[-1]
|
||
if not isinstance(message, HumanMessage):
|
||
return None
|
||
additional_kwargs = getattr(message, "additional_kwargs", {}) or {}
|
||
if any(additional_kwargs.get(key) for key in ("files", "uploaded_files", "attachments", "prompt_prefix")):
|
||
return None
|
||
content = getattr(message, "content", None)
|
||
if isinstance(content, str):
|
||
return content
|
||
if not isinstance(content, list):
|
||
return None
|
||
parts: list[str] = []
|
||
for item in content:
|
||
if isinstance(item, str):
|
||
parts.append(item)
|
||
continue
|
||
if not isinstance(item, Mapping):
|
||
return None
|
||
item_type = str(item.get("type") or "text")
|
||
text = item.get("text")
|
||
if item_type not in {"text", "input_text"} or not isinstance(text, str):
|
||
return None
|
||
parts.append(text)
|
||
return "".join(parts)
|
||
|
||
|
||
async def _find_fixed_answer(
|
||
*,
|
||
body: Any,
|
||
request: Request,
|
||
graph_input: dict[str, Any],
|
||
run_scope: str,
|
||
) -> dict[str, Any] | None:
|
||
"""Look up an enabled exact match, falling back to the model on errors."""
|
||
|
||
if run_scope != RUN_SCOPE_EXTERNAL_ROOT or getattr(body, "command", None) is not None:
|
||
return None
|
||
assistant_id = str(getattr(body, "assistant_id", None) or _DEFAULT_ASSISTANT_ID)
|
||
if assistant_id in {"canvas_agent", "ai_writing"}:
|
||
return None
|
||
context = getattr(body, "context", None) or {}
|
||
if isinstance(context, Mapping) and context.get("writing_mode"):
|
||
return None
|
||
question = _plain_fixed_question(graph_input)
|
||
if question is None:
|
||
return None
|
||
store = getattr(request.app.state, "fixed_question_store", None)
|
||
if store is None:
|
||
return None
|
||
try:
|
||
return await store.find_enabled(question)
|
||
except Exception:
|
||
# Configuration storage must never make normal chat unavailable.
|
||
logger.warning("Fixed-question lookup failed; continuing with model", exc_info=True)
|
||
return None
|
||
|
||
|
||
_DEFAULT_ASSISTANT_ID = "lead_agent"
|
||
|
||
# Built-in LangGraph graph IDs registered in langgraph.json.
|
||
# These are NOT stored in the agents DB and must bypass the custom-agent access check.
|
||
_BUILTIN_GRAPH_IDS: frozenset[str] = frozenset({"lead_agent", "canvas_agent", "ai_writing"})
|
||
|
||
_RAG_CITATION_PROMPT = """当前对话已开启参考文献模式。
|
||
如果本轮调用了搜索、知识库、检索类技能或工具，请基于本轮所有检索结果整体回答。
|
||
如果你已经知道需要多个检索词，请优先在同一步里并行调用检索工具，等这些检索都返回后再回答，不要查一个检索词就先写一版答案。
|
||
你必须把同一轮内多次检索得到的资料视为同一个 referenceBatch，按系统稍后提供的合并批次从 [1] 开始连续编号，不要让每个检索词各自从 [1] 重新编号。
|
||
回答正文中的关键结论必须标注对应编号，例如 [1]、[2]，多个来源可写为 [1][3][5]。
|
||
不要引用不存在的编号，不要使用技能内部自带的编号，不要把参考文献完整列表堆在正文末尾。
|
||
如果没有任何可用检索结果，请直接说明未检索到可靠参考资料。"""
|
||
|
||
|
||
def _apply_classic_rag_citation_prompt(
|
||
graph_input: dict[str, Any],
|
||
*,
|
||
enabled: bool,
|
||
knowledge_retrieval_mode: str,
|
||
) -> dict[str, Any]:
|
||
"""Inject numbered-reference instructions for classic RAG only.
|
||
|
||
Smart reasoning consumes structured evidence inside its own ReAct loop and
|
||
intentionally answers without a mandatory ``[1]``/``[2]`` contract or a
|
||
generated bibliography. Keeping this policy in one helper prevents a
|
||
stale thread RAG flag from re-enabling the classic prompt.
|
||
"""
|
||
|
||
if not enabled or knowledge_retrieval_mode == "smart_reasoning":
|
||
return graph_input
|
||
return _append_invisible_prompt(graph_input, _RAG_CITATION_PROMPT)
|
||
|
||
|
||
def _prepend_prompt_prefix(graph_input: dict[str, Any], prefix: str) -> dict[str, Any]:
|
||
"""Return ``graph_input`` with the prefix *tagged* onto its first HumanMessage.
|
||
|
||
The prefix is stored in ``additional_kwargs['prompt_prefix']`` rather than
|
||
concatenated into ``content``. This keeps the persisted/displayed message
|
||
clean — the user never sees the prefix — while ``PromptPrefixMiddleware``
|
||
injects it into the content of the request actually sent to the LLM.
|
||
|
||
Leaves the input unchanged when there are no messages or the first message
|
||
is not a human turn.
|
||
"""
|
||
if not prefix:
|
||
return graph_input
|
||
messages = graph_input.get("messages")
|
||
if not isinstance(messages, list) or not messages:
|
||
return graph_input
|
||
new_messages = list(messages)
|
||
for idx, msg in enumerate(new_messages):
|
||
if isinstance(msg, HumanMessage):
|
||
extra = dict(getattr(msg, "additional_kwargs", None) or {})
|
||
extra["prompt_prefix"] = prefix
|
||
new_messages[idx] = msg.model_copy(update={"additional_kwargs": extra})
|
||
break
|
||
return {**graph_input, "messages": new_messages}
|
||
|
||
|
||
def _prompt_prefix_opted_in(body: Any) -> bool:
|
||
"""Detect whether the caller asked for the prompt prefix to be applied."""
|
||
context = getattr(body, "context", None)
|
||
if isinstance(context, Mapping) and context.get("prompt_prefix_enabled"):
|
||
return True
|
||
metadata = getattr(body, "metadata", None)
|
||
if isinstance(metadata, Mapping) and metadata.get("prompt_prefix_enabled"):
|
||
return True
|
||
return False
|
||
|
||
|
||
def _explicit_rag_mode_requested(body: Any) -> bool:
|
||
context = getattr(body, "context", None)
|
||
if isinstance(context, Mapping) and context.get("rag_mode_enabled"):
|
||
return True
|
||
metadata = getattr(body, "metadata", None)
|
||
if isinstance(metadata, Mapping) and (metadata.get("rag_mode_enabled") or metadata.get("rag_mode")):
|
||
return True
|
||
return False
|
||
|
||
|
||
def _rag_mode_requested(body: Any) -> bool:
|
||
return llmwiki_selection_requested(body) or _explicit_rag_mode_requested(body)
|
||
|
||
|
||
def _effective_rag_mode_after_llmwiki(
|
||
*,
|
||
body: Any,
|
||
rag: LlmWikiRagContext,
|
||
inherited: bool,
|
||
) -> bool:
|
||
"""Resolve RAG mode without treating an observability trace as evidence.
|
||
|
||
A disabled trace is deliberately retained when the caller explicitly
|
||
clears both knowledge-space selections. That trace must not reactivate a
|
||
thread's citation prompt or persist ``rag_mode=true`` again. An explicit
|
||
generic RAG flag still wins, while any resolved ordinary/assistant space
|
||
enables the strict knowledge mode.
|
||
"""
|
||
|
||
if rag.knowledge_base_ids or rag.assistant_knowledge_base_ids:
|
||
return True
|
||
if llmwiki_selection_declared(body) and not _explicit_rag_mode_requested(body):
|
||
return False
|
||
return inherited
|
||
|
||
|
||
def _retrieval_rag_metadata_flags(
|
||
*,
|
||
knowledge_retrieval_mode: str,
|
||
knowledge_scope_active: bool,
|
||
selection_declared: bool,
|
||
explicit_rag_requested: bool,
|
||
) -> dict[str, bool]:
|
||
"""Return the authoritative classic-reference flags for persisted state."""
|
||
|
||
if knowledge_retrieval_mode == "smart_reasoning":
|
||
return {"rag_mode_enabled": False, "rag_mode": False}
|
||
if knowledge_scope_active:
|
||
return {"rag_mode_enabled": True, "rag_mode": True}
|
||
if selection_declared and not explicit_rag_requested:
|
||
return {"rag_mode_enabled": False, "rag_mode": False}
|
||
return {}
|
||
|
||
|
||
def _sync_retrieval_rag_metadata(
|
||
*,
|
||
run_metadata: dict[str, Any],
|
||
record_metadata: dict[str, Any],
|
||
thread_metadata_update: dict[str, Any],
|
||
knowledge_retrieval_mode: str,
|
||
knowledge_scope_active: bool,
|
||
selection_declared: bool,
|
||
explicit_rag_requested: bool,
|
||
) -> dict[str, bool]:
|
||
"""Keep run, record and durable thread reference-mode flags identical."""
|
||
|
||
flags = _retrieval_rag_metadata_flags(
|
||
knowledge_retrieval_mode=knowledge_retrieval_mode,
|
||
knowledge_scope_active=knowledge_scope_active,
|
||
selection_declared=selection_declared,
|
||
explicit_rag_requested=explicit_rag_requested,
|
||
)
|
||
thread_metadata_update.update(flags)
|
||
run_metadata.update(flags)
|
||
record_metadata.update(flags)
|
||
return flags
|
||
|
||
|
||
def _thread_rag_mode_enabled(record: Any) -> bool:
|
||
if not isinstance(record, Mapping):
|
||
return False
|
||
metadata = record.get("metadata")
|
||
return isinstance(metadata, Mapping) and bool(metadata.get("rag_mode_enabled") or metadata.get("rag_mode"))
|
||
|
||
|
||
def _append_invisible_prompt(graph_input: dict[str, Any], prompt: str) -> dict[str, Any]:
|
||
messages = graph_input.get("messages")
|
||
if not isinstance(messages, list) or not messages:
|
||
return graph_input
|
||
new_messages = list(messages)
|
||
for idx, msg in enumerate(new_messages):
|
||
if not isinstance(msg, HumanMessage):
|
||
continue
|
||
extra = dict(getattr(msg, "additional_kwargs", None) or {})
|
||
existing = str(extra.get("prompt_prefix") or "").strip()
|
||
extra["prompt_prefix"] = f"{existing}\n\n{prompt}".strip() if existing else prompt
|
||
new_messages[idx] = msg.model_copy(update={"additional_kwargs": extra})
|
||
break
|
||
return {**graph_input, "messages": new_messages}
|
||
|
||
|
||
# Whitelist of run-context keys that the langgraph-compat layer forwards from
|
||
# ``body.context`` into the run config. ``config["context"]`` exists in
|
||
# LangGraph >=0.6, but these values must be written to both ``configurable``
|
||
# (for legacy ``_get_runtime_config`` consumers) and ``context`` because
|
||
# LangGraph >=1.1.9 no longer makes ``ToolRuntime.context`` fall back to
|
||
# ``configurable`` for consumers like ``setup_agent``.
|
||
_CONTEXT_CONFIGURABLE_KEYS: frozenset[str] = frozenset(
|
||
{
|
||
"agent_id",
|
||
"model_name",
|
||
"mode",
|
||
"thinking_enabled",
|
||
# 强制关闭思考(内网 vLLM/Qwen 模型只声明 supports_thinking 时,单靠
|
||
# thinking_enabled=False 不会注入 enable_thinking=false)。透传到 run config
|
||
# 后由 lead agent 交给 create_chat_model(force_disable_thinking=...)。
|
||
"thinking_force_disabled",
|
||
"reasoning_effort",
|
||
"rag_mode_enabled",
|
||
"knowledge_rag_enabled",
|
||
"llmwiki_knowledge_base_ids",
|
||
"assistant_knowledge_base_ids",
|
||
"knowledge_space_strict",
|
||
"knowledge_retrieval_mode",
|
||
"writing_mode",
|
||
"writing_artifact_path",
|
||
# 深度研究工作台「开始撰写」：走写作模式内联管线，不建 job。
|
||
"workbench_write_now",
|
||
"workbench_session_id",
|
||
"writing_report_config",
|
||
"force_no_materials",
|
||
# 收集员当场指定的检索技能（报告结构「检索来源」）。
|
||
"extra_skills",
|
||
"canvas_artifact_title",
|
||
"canvas_artifact_type",
|
||
"is_plan_mode",
|
||
"subagent_enabled",
|
||
"max_concurrent_subagents",
|
||
"agent_name",
|
||
# 页面研报设计师由前端随消息带入当前界面明暗模式；需要进入 agent
|
||
# 提示词及 ToolRuntime.context，才能选择同色系模板。
|
||
"presentation_theme",
|
||
"is_bootstrap",
|
||
"is_writing_setup",
|
||
"is_report_structure_setup",
|
||
"is_gongwen_template_setup",
|
||
"valid_article_types",
|
||
"valid_retrieval_skills",
|
||
"writing_setup_quick",
|
||
"writing_sample_context",
|
||
"excluded_tools",
|
||
# Internal role policies may opt into a fail-closed tool allowlist.
|
||
# Ordinary clients omit it and retain the existing tool resolution.
|
||
"allowed_tools",
|
||
"skill_stop_names",
|
||
"skill_stop_message",
|
||
# 记忆开关:必须在白名单里才会从 body.context 透传到 run config,
|
||
# 否则 agent 端 cfg.get(...) 永远取默认值。圆桌等临时任务靠这两个键
|
||
# 关掉个人记忆召回 / builtin 记忆注入(见 multi_agent._run_payload)。
|
||
"memory_recall_disabled",
|
||
"memory_injection_enabled",
|
||
# 任务工作区 / 岗位会商把左侧任务卡的快照放在这里。它必须进入
|
||
# ``runtime.context``，TaskExternalContextMiddleware 才能在每次模型调用
|
||
# 前以隐藏消息注入任务名称、描述和方向；未白名单会被静默丢弃，模型只能
|
||
# 看到用户说的“这个任务”。
|
||
"task_external_context",
|
||
# 岗位会商下游席位按需读取前序席位完整交付时使用的运行期数据。
|
||
"roundtable_peer_deliveries",
|
||
# 岗位会商席位 / 方案总结：每轮 outputs 下 write_file 上限（行动规划不传此键）。
|
||
"position_artifact_cap",
|
||
# 会商专用文件流式兼容。只由会商 run 写入；普通对话不带此键，仍只看全局开关。
|
||
"roundtable_write_file_compatibility",
|
||
# 上下文压缩开关:由聊天框下方的按钮控制(默认关闭),透传到 run config 后
|
||
# 在 _build_middlewares 里决定是否挂载 SummarizationMiddleware。
|
||
"summarization_enabled",
|
||
}
|
||
)
|
||
|
||
_KNOWLEDGE_SPACE_EXTERNAL_RETRIEVAL_TOOLS: frozenset[str] = frozenset(
|
||
{
|
||
"web_search",
|
||
"web_fetch",
|
||
"image_search",
|
||
"browser_fetch_page",
|
||
"browser_act",
|
||
"knowledge_search",
|
||
"notebook_search",
|
||
"tool_search",
|
||
"invoke_acp_agent",
|
||
"agent_orchestration",
|
||
"task",
|
||
}
|
||
)
|
||
|
||
|
||
def knowledge_space_strict_excluded_tools(
|
||
existing: list[str] | None,
|
||
*,
|
||
app_config: Any | None = None,
|
||
) -> list[str]:
|
||
"""Merge fail-closed external retrieval exclusions for selected spaces."""
|
||
|
||
blocked = set(_KNOWLEDGE_SPACE_EXTERNAL_RETRIEVAL_TOOLS)
|
||
resolved = app_config or get_app_config()
|
||
for tool in getattr(resolved, "tools", []) or []:
|
||
if str(getattr(tool, "group", "") or "").strip().lower() in {"web", "browser"}:
|
||
blocked.add(str(getattr(tool, "name", "") or "").strip())
|
||
return list(dict.fromkeys([*(existing or []), *(name for name in sorted(blocked) if name)]))
|
||
|
||
|
||
def apply_knowledge_space_strict_scope(
|
||
context: dict[str, Any],
|
||
*,
|
||
selection_requested: bool,
|
||
knowledge_base_ids: list[str],
|
||
assistant_knowledge_base_ids: list[str],
|
||
app_config: Any | None = None,
|
||
) -> bool:
|
||
"""Fail closed whenever a run declares a selected knowledge space.
|
||
|
||
``selection_requested`` remains authoritative even when retrieval returns no
|
||
authorized ids (for example, while an upstream index is unavailable). A
|
||
failed lookup must never silently reopen Web/browser/MCP-style retrieval.
|
||
"""
|
||
|
||
if not (
|
||
selection_requested
|
||
or knowledge_base_ids
|
||
or assistant_knowledge_base_ids
|
||
):
|
||
return False
|
||
raw_excluded = context.get("excluded_tools")
|
||
if isinstance(raw_excluded, str):
|
||
excluded = [raw_excluded] if raw_excluded.strip() else []
|
||
elif isinstance(raw_excluded, (list, tuple, set)):
|
||
excluded = [str(item) for item in raw_excluded if str(item).strip()]
|
||
else:
|
||
excluded = []
|
||
context["knowledge_space_strict"] = True
|
||
context["excluded_tools"] = knowledge_space_strict_excluded_tools(
|
||
excluded,
|
||
app_config=app_config,
|
||
)
|
||
return True
|
||
|
||
|
||
def enforce_knowledge_space_strict_run_config(
|
||
config: dict[str, Any],
|
||
*,
|
||
knowledge_base_ids: list[str],
|
||
assistant_knowledge_base_ids: list[str],
|
||
excluded_tools: list[str],
|
||
app_config: Any | None = None,
|
||
) -> None:
|
||
"""Overwrite both runtime scopes with the server-validated selection.
|
||
|
||
``merge_run_context_overrides`` intentionally uses ``setdefault`` for
|
||
ordinary client preferences. Security policy is different: a caller must
|
||
not pre-seed ``knowledge_space_strict=false``, alternate ids, or an empty
|
||
exclusion list in ``config.context``/``configurable`` to reopen retrieval.
|
||
"""
|
||
|
||
for scope in ("configurable", "context"):
|
||
target = config.setdefault(scope, {})
|
||
if not isinstance(target, dict):
|
||
continue
|
||
raw_target_excluded = target.get("excluded_tools")
|
||
if isinstance(raw_target_excluded, str):
|
||
target_excluded = (
|
||
[raw_target_excluded] if raw_target_excluded.strip() else []
|
||
)
|
||
elif isinstance(raw_target_excluded, (list, tuple, set)):
|
||
target_excluded = [
|
||
str(item) for item in raw_target_excluded if str(item).strip()
|
||
]
|
||
else:
|
||
target_excluded = []
|
||
target["knowledge_space_strict"] = True
|
||
target["llmwiki_knowledge_base_ids"] = list(knowledge_base_ids)
|
||
target["assistant_knowledge_base_ids"] = list(
|
||
assistant_knowledge_base_ids
|
||
)
|
||
target["excluded_tools"] = knowledge_space_strict_excluded_tools(
|
||
[*target_excluded, *excluded_tools],
|
||
app_config=app_config,
|
||
)
|
||
|
||
|
||
def merge_run_context_overrides(config: dict[str, Any], context: Mapping[str, Any] | None) -> None:
|
||
"""Merge whitelisted keys from ``body.context`` into both ``config['configurable']``
|
||
and ``config['context']`` so they are visible to legacy configurable readers and
|
||
to LangGraph ``ToolRuntime.context`` consumers (e.g. the ``setup_agent`` tool —
|
||
see issue #2677)."""
|
||
if not context:
|
||
return
|
||
configurable = config.setdefault("configurable", {})
|
||
runtime_context = config.setdefault("context", {})
|
||
for key in _CONTEXT_CONFIGURABLE_KEYS:
|
||
if key in context:
|
||
if isinstance(configurable, dict):
|
||
configurable.setdefault(key, context[key])
|
||
if isinstance(runtime_context, dict):
|
||
runtime_context.setdefault(key, context[key])
|
||
|
||
|
||
def enforce_knowledge_retrieval_mode_run_config(
|
||
config: dict[str, Any],
|
||
mode: str,
|
||
) -> None:
|
||
"""Write the server-validated retrieval mode into both runtime scopes."""
|
||
|
||
smart_tool_names = {
|
||
"ordinary_knowledge_search",
|
||
"ordinary_chunk_regex_search",
|
||
"ordinary_document_chunks",
|
||
"ordinary_wiki_search",
|
||
"ordinary_wiki_read_page",
|
||
}
|
||
classic_tool_names = set(CLASSIC_LLMWIKI_TOOL_NAMES)
|
||
for scope in ("configurable", "context"):
|
||
target = config.setdefault(scope, {})
|
||
if isinstance(target, dict):
|
||
target["knowledge_retrieval_mode"] = mode
|
||
if mode == "smart_reasoning":
|
||
# Smart mode keeps structured source traces but deliberately
|
||
# does not load the classic numbered-citation prompt.
|
||
target["rag_mode_enabled"] = False
|
||
target["knowledge_rag_enabled"] = False
|
||
# The mode is useful only when the five server-owned read tools
|
||
# are mounted. A caller may pre-seed ``config.context``; do not
|
||
# let that untrusted layer disable or exclude the validated
|
||
# retrieval surface.
|
||
target["disable_tools"] = False
|
||
raw_excluded = target.get("excluded_tools")
|
||
if isinstance(raw_excluded, str):
|
||
excluded = [raw_excluded] if raw_excluded.strip() else []
|
||
elif isinstance(raw_excluded, (list, tuple, set)):
|
||
excluded = [
|
||
str(item)
|
||
for item in raw_excluded
|
||
if str(item).strip()
|
||
]
|
||
else:
|
||
excluded = []
|
||
target["excluded_tools"] = list(
|
||
dict.fromkeys(
|
||
[
|
||
*(name for name in excluded if name not in smart_tool_names),
|
||
*sorted(classic_tool_names),
|
||
]
|
||
)
|
||
)
|
||
else:
|
||
raw_excluded = target.get("excluded_tools")
|
||
if isinstance(raw_excluded, str):
|
||
excluded = [raw_excluded] if raw_excluded.strip() else []
|
||
elif isinstance(raw_excluded, (list, tuple, set)):
|
||
excluded = [str(item) for item in raw_excluded if str(item).strip()]
|
||
else:
|
||
excluded = []
|
||
ordinary_ids = target.get("llmwiki_knowledge_base_ids")
|
||
assistant_ids = target.get("assistant_knowledge_base_ids")
|
||
has_ordinary = isinstance(ordinary_ids, list) and bool(ordinary_ids)
|
||
has_assistant = isinstance(assistant_ids, list) and bool(assistant_ids)
|
||
enabled_classic_tools: set[str] = set()
|
||
if has_ordinary or has_assistant:
|
||
enabled_classic_tools.add("llmwiki_search")
|
||
if has_ordinary:
|
||
enabled_classic_tools.update(
|
||
{"llmwiki_browse_wiki_index", "llmwiki_read_wiki_page"}
|
||
)
|
||
target["excluded_tools"] = list(
|
||
dict.fromkeys(
|
||
[
|
||
*(
|
||
name
|
||
for name in excluded
|
||
if name not in enabled_classic_tools
|
||
),
|
||
*sorted(smart_tool_names),
|
||
*sorted(classic_tool_names - enabled_classic_tools),
|
||
]
|
||
)
|
||
)
|
||
metadata = config.get("metadata")
|
||
if mode == "smart_reasoning" and isinstance(metadata, dict):
|
||
# Run-config metadata is observable by tracing/stream consumers even
|
||
# though the model reads context/configurable. Keep all three views
|
||
# consistent so smart mode cannot appear as classic reference mode.
|
||
metadata["rag_mode_enabled"] = False
|
||
metadata["rag_mode"] = False
|
||
|
||
|
||
async def resolve_knowledge_retrieval_mode(request: Request, body: Any) -> str:
|
||
"""Validate the explicit per-run mode against the database-backed switch."""
|
||
|
||
context = getattr(body, "context", None)
|
||
raw = context.get("knowledge_retrieval_mode") if isinstance(context, Mapping) else None
|
||
if raw is None or (isinstance(raw, str) and not raw.strip()):
|
||
return "classic"
|
||
if not isinstance(raw, str) or raw.strip().lower() not in {
|
||
"classic",
|
||
"smart_reasoning",
|
||
}:
|
||
raise HTTPException(
|
||
status_code=422,
|
||
detail={
|
||
"code": "INVALID_KNOWLEDGE_RETRIEVAL_MODE",
|
||
"message": "knowledge_retrieval_mode 必须为 classic 或 smart_reasoning。",
|
||
},
|
||
)
|
||
mode = raw.strip().lower()
|
||
if mode == "classic":
|
||
return mode
|
||
store = getattr(request.app.state, "knowledge_retrieval_settings_store", None)
|
||
if store is None:
|
||
raise HTTPException(
|
||
status_code=503,
|
||
detail={
|
||
"code": "KNOWLEDGE_RETRIEVAL_SETTINGS_UNAVAILABLE",
|
||
"message": "智能推理检索设置服务当前不可用。",
|
||
},
|
||
)
|
||
try:
|
||
enabled = bool(await store.get())
|
||
except Exception as exc:
|
||
logger.warning("Failed to read database-backed knowledge retrieval setting", exc_info=True)
|
||
raise HTTPException(
|
||
status_code=503,
|
||
detail={
|
||
"code": "KNOWLEDGE_RETRIEVAL_SETTINGS_UNAVAILABLE",
|
||
"message": "智能推理检索设置服务当前不可用。",
|
||
},
|
||
) from exc
|
||
if not enabled:
|
||
raise HTTPException(
|
||
status_code=409,
|
||
detail={
|
||
"code": "SMART_REASONING_DISABLED",
|
||
"message": "管理员尚未开启普通知识库智能推理检索。",
|
||
},
|
||
)
|
||
return mode
|
||
|
||
|
||
_REASONING_EFFORT_ALIASES = {
|
||
"minimal": "low",
|
||
"low": "low",
|
||
"medium": "medium",
|
||
"high": "xhigh",
|
||
"xhigh": "xhigh",
|
||
}
|
||
|
||
|
||
def _normalize_reasoning_effort(value: Any) -> str:
|
||
"""Normalize the public selector to Qwen's low/medium/xhigh contract."""
|
||
|
||
if value is None or (isinstance(value, str) and not value.strip()):
|
||
return "xhigh"
|
||
if not isinstance(value, str):
|
||
raise HTTPException(status_code=422, detail="reasoning_effort must be low, medium, or xhigh")
|
||
normalized = _REASONING_EFFORT_ALIASES.get(value.strip().lower())
|
||
if normalized is None:
|
||
raise HTTPException(status_code=422, detail="reasoning_effort must be low, medium, or xhigh")
|
||
return normalized
|
||
|
||
|
||
def enforce_reasoning_effort_run_config(
|
||
config: dict[str, Any],
|
||
*,
|
||
capability_enabled: bool,
|
||
model_supports_thinking: bool,
|
||
default_reasoning_effort: str = "xhigh",
|
||
) -> str | None:
|
||
"""Apply the database-backed reasoning policy to both runtime containers.
|
||
|
||
This intentionally runs *after* the client ``config`` and ``context`` have
|
||
been merged. A caller therefore cannot bypass the capability gate by
|
||
pre-seeding ``config.configurable.reasoning_effort``.
|
||
"""
|
||
|
||
configurable = config.setdefault("configurable", {})
|
||
runtime_context = config.setdefault("context", {})
|
||
effective: dict[str, Any] = {}
|
||
if isinstance(configurable, Mapping):
|
||
effective.update(configurable)
|
||
if isinstance(runtime_context, Mapping):
|
||
effective.update(runtime_context)
|
||
|
||
thinking_enabled = bool(effective.get("thinking_enabled", True))
|
||
if capability_enabled and model_supports_thinking and thinking_enabled:
|
||
# The selected model's persisted default is authoritative. A per-thread
|
||
# value from an older browser must not override the administrator's
|
||
# globally shared model policy.
|
||
normalized = _normalize_reasoning_effort(default_reasoning_effort)
|
||
else:
|
||
normalized = None
|
||
|
||
for target in (configurable, runtime_context):
|
||
if not isinstance(target, dict):
|
||
continue
|
||
if normalized is None:
|
||
target.pop("reasoning_effort", None)
|
||
else:
|
||
target["reasoning_effort"] = normalized
|
||
return normalized
|
||
|
||
|
||
async def _reasoning_effort_policy(request: Request, model_name: str) -> tuple[bool, str]:
|
||
"""Read one model capability and its shared default, failing closed."""
|
||
|
||
store = getattr(request.app.state, "model_capability_store", None)
|
||
if store is None:
|
||
return False, "xhigh"
|
||
try:
|
||
row = await store.get_capability(model_name)
|
||
except Exception:
|
||
logger.warning(
|
||
"Failed to read reasoning-effort capability for model %s",
|
||
sanitize_log_param(model_name),
|
||
exc_info=True,
|
||
)
|
||
return False, "xhigh"
|
||
if not row:
|
||
return False, "xhigh"
|
||
default = str(row.get("default_reasoning_effort") or "xhigh")
|
||
if default not in {"low", "medium", "xhigh"}:
|
||
default = "xhigh"
|
||
return bool(row.get("supports_reasoning_effort", False)), default
|
||
|
||
|
||
def inject_validated_analysis_project_context(config: dict[str, Any], context: Mapping[str, Any] | None) -> None:
|
||
"""Inject server-derived project policy outside the public context whitelist."""
|
||
if not context or not isinstance(context.get("analysis_project"), dict):
|
||
return
|
||
project_policy = {
|
||
"analysis_project": context["analysis_project"],
|
||
"solution_requirements_confirmed": bool(context.get("solution_requirements_confirmed")),
|
||
"solution_force_generate": bool(context.get("solution_force_generate")),
|
||
"solution_intent": context.get("solution_intent") if isinstance(context.get("solution_intent"), dict) else None,
|
||
}
|
||
project_memory_scope = context.get("project_memory_scope")
|
||
if isinstance(project_memory_scope, dict):
|
||
project_policy.update(
|
||
{
|
||
"project_memory_scope": project_memory_scope,
|
||
"project_memory": context.get("project_memory")
|
||
if isinstance(context.get("project_memory"), dict)
|
||
else None,
|
||
# Complete isolation from the legacy/global memory path.
|
||
"memory_recall_disabled": True,
|
||
"memory_injection_enabled": False,
|
||
}
|
||
)
|
||
for scope in ("configurable", "context"):
|
||
target = config.setdefault(scope, {})
|
||
if isinstance(target, dict):
|
||
# Assignment is intentional: these values were derived from the
|
||
# immutable project binding/checkpoint and override request config.
|
||
target.update(project_policy)
|
||
if isinstance(project_memory_scope, dict):
|
||
raw_excluded = target.get("excluded_tools")
|
||
if isinstance(raw_excluded, str):
|
||
excluded = [raw_excluded] if raw_excluded.strip() else []
|
||
elif isinstance(raw_excluded, (list, tuple, set)):
|
||
excluded = [str(item) for item in raw_excluded if str(item).strip()]
|
||
else:
|
||
excluded = []
|
||
target["excluded_tools"] = list(
|
||
dict.fromkeys(
|
||
[
|
||
*excluded,
|
||
"memory",
|
||
"hindsight_recall",
|
||
"hindsight_reflect",
|
||
"hindsight_retain",
|
||
]
|
||
)
|
||
)
|
||
|
||
|
||
def resolve_agent_factory(assistant_id: str | None):
|
||
"""Resolve the agent factory callable from config.
|
||
|
||
Built-in graphs (``lead_agent``, ``canvas_agent``, ``ai_writing``) are dispatched
|
||
directly to their own factory. Custom agents share the ``lead_agent`` factory
|
||
with ``agent_id`` injected into configurable/context by :func:`build_run_config`.
|
||
"""
|
||
if assistant_id == "canvas_agent":
|
||
from deerflow.agents.canvas_agent.agent import make_canvas_agent
|
||
|
||
return make_canvas_agent
|
||
|
||
if assistant_id == "ai_writing":
|
||
from deerflow.agents.ai_writing.graph import make_ai_writing_graph
|
||
|
||
return make_ai_writing_graph
|
||
|
||
from deerflow.agents.lead_agent.agent import make_lead_agent
|
||
|
||
return make_lead_agent
|
||
|
||
|
||
def build_run_config(
|
||
thread_id: str,
|
||
request_config: dict[str, Any] | None,
|
||
metadata: dict[str, Any] | None,
|
||
*,
|
||
assistant_id: str | None = None,
|
||
) -> dict[str, Any]:
|
||
"""Build a RunnableConfig dict for the agent.
|
||
|
||
When *assistant_id* refers to a custom agent (anything other than
|
||
``"lead_agent"`` / ``None``), the id is forwarded as ``agent_id`` in
|
||
whichever runtime options container is active: ``context`` for
|
||
LangGraph >= 0.6.0 requests, otherwise ``configurable``.
|
||
``make_lead_agent`` reads this key to load the matching
|
||
``agents/<id>/SOUL.md`` and per-agent config — without it the agent
|
||
silently runs as the default lead agent.
|
||
|
||
This mirrors the channel manager's ``_resolve_run_params`` logic so that
|
||
the LangGraph Platform-compatible HTTP API and the IM channel path behave
|
||
identically.
|
||
"""
|
||
config: dict[str, Any] = {"recursion_limit": 250}
|
||
if request_config:
|
||
# LangGraph >= 0.6.0 introduced ``context`` as the preferred way to
|
||
# pass thread-level data and rejects requests that include both
|
||
# ``configurable`` and ``context``. If the caller already sends
|
||
# ``context``, honour it and skip our own ``configurable`` dict.
|
||
if "context" in request_config:
|
||
if "configurable" in request_config:
|
||
logger.warning(
|
||
"build_run_config: client sent both 'context' and 'configurable'; preferring 'context' (LangGraph >= 0.6.0). thread_id=%s, caller_configurable keys=%s",
|
||
thread_id,
|
||
list(request_config.get("configurable", {}).keys()),
|
||
)
|
||
context_value = request_config["context"]
|
||
if context_value is None:
|
||
context = {}
|
||
elif isinstance(context_value, Mapping):
|
||
context = dict(context_value)
|
||
else:
|
||
raise ValueError("request config 'context' must be a mapping or null.")
|
||
config["context"] = context
|
||
else:
|
||
configurable = {"thread_id": thread_id}
|
||
configurable.update(request_config.get("configurable", {}))
|
||
config["configurable"] = configurable
|
||
for k, v in request_config.items():
|
||
if k not in ("configurable", "context"):
|
||
config[k] = v
|
||
else:
|
||
config["configurable"] = {"thread_id": thread_id}
|
||
|
||
# Inject custom agent id when the caller specified a non-default assistant.
|
||
# Built-in graphs (lead_agent, canvas_agent) are dispatched via their own
|
||
# factory and must NOT receive an agent_id override.
|
||
if assistant_id and assistant_id not in _BUILTIN_GRAPH_IDS:
|
||
normalized = assistant_id.strip()
|
||
if not normalized or not re.fullmatch(r"[A-Za-z0-9_-]+", normalized):
|
||
raise ValueError(f"Invalid assistant_id {assistant_id!r}: must contain only letters, digits, hyphens, and underscores.")
|
||
if "configurable" in config:
|
||
target = config["configurable"]
|
||
elif "context" in config:
|
||
target = config["context"]
|
||
else:
|
||
target = config.setdefault("configurable", {})
|
||
if target is not None and "agent_id" not in target:
|
||
target["agent_id"] = normalized
|
||
if metadata:
|
||
config.setdefault("metadata", {}).update(metadata)
|
||
# Chat clients cannot override the administrator's deployment-wide budget.
|
||
# Other built-in graphs (writing/canvas) retain their own execution policy.
|
||
if assistant_id not in _BUILTIN_GRAPH_IDS or assistant_id == "lead_agent":
|
||
from deerflow.config.system_settings import get_conversation_context_settings
|
||
|
||
config["recursion_limit"] = get_conversation_context_settings().max_run_steps
|
||
return config
|
||
|
||
|
||
def _collect_requested_agent_ids(body: Any) -> set[str]:
|
||
"""Collect agent ids from assistant_id/config/context for access checks."""
|
||
agent_ids: set[str] = set()
|
||
if body.assistant_id and body.assistant_id not in _BUILTIN_GRAPH_IDS:
|
||
agent_ids.add(str(body.assistant_id))
|
||
|
||
def collect(container: Mapping[str, Any] | None) -> None:
|
||
if not container:
|
||
return
|
||
for key in ("agent_id", "agent_name"):
|
||
value = container.get(key)
|
||
if value:
|
||
agent_ids.add(str(value))
|
||
|
||
context = getattr(body, "context", None)
|
||
if isinstance(context, Mapping):
|
||
collect(context)
|
||
|
||
config = body.config or {}
|
||
if isinstance(config, Mapping):
|
||
nested_context = config.get("context")
|
||
nested_configurable = config.get("configurable")
|
||
if isinstance(nested_context, Mapping):
|
||
collect(nested_context)
|
||
if isinstance(nested_configurable, Mapping):
|
||
collect(nested_configurable)
|
||
|
||
return agent_ids
|
||
|
||
|
||
def _runtime_config_values(config: Mapping[str, Any]) -> dict[str, Any]:
|
||
"""Mirror the lead agent's configurable/context precedence exactly."""
|
||
|
||
configurable = config.get("configurable")
|
||
runtime_context = config.get("context")
|
||
values = dict(configurable) if isinstance(configurable, Mapping) else {}
|
||
if isinstance(runtime_context, Mapping):
|
||
values.update(runtime_context)
|
||
return values
|
||
|
||
|
||
def _resolve_runtime_model_name(config: Mapping[str, Any], app_config: Any) -> str:
|
||
"""Resolve the model the lead agent will use from its final RunnableConfig.
|
||
|
||
Request model overrides win over custom-agent defaults, and an unknown
|
||
explicit model falls back to the global default just like
|
||
``lead_agent._resolve_model_name``. When no model override exists, the
|
||
effective runtime agent id supplies its configured model.
|
||
"""
|
||
|
||
default_model_name = app_config.models[0].name if app_config.models else None
|
||
if not default_model_name:
|
||
raise ValueError("No chat models are configured. Please configure at least one model in config.yaml.")
|
||
|
||
runtime = _runtime_config_values(config)
|
||
requested_model_name = str(runtime.get("model_name") or runtime.get("model") or "").strip()
|
||
if requested_model_name:
|
||
if app_config.get_model_config(requested_model_name):
|
||
return requested_model_name
|
||
logger.warning(
|
||
"Requested model '%s' not found; falling back to default model '%s'.",
|
||
requested_model_name,
|
||
default_model_name,
|
||
)
|
||
return default_model_name
|
||
|
||
raw_agent_id = runtime.get("agent_id") or runtime.get("agent_name")
|
||
try:
|
||
agent_id = validate_agent_id(raw_agent_id)
|
||
except ValueError:
|
||
logger.debug("Ignoring invalid agent id during model resolution: %s", raw_agent_id)
|
||
agent_id = None
|
||
if agent_id:
|
||
try:
|
||
agent_config = load_agent_config(agent_id)
|
||
except FileNotFoundError:
|
||
agent_config = None
|
||
except Exception:
|
||
logger.debug("Failed to load agent config for model resolution: %s", agent_id, exc_info=True)
|
||
agent_config = None
|
||
agent_model_name = str(getattr(agent_config, "model", "") or "").strip()
|
||
if agent_model_name:
|
||
if app_config.get_model_config(agent_model_name):
|
||
return agent_model_name
|
||
logger.warning(
|
||
"Agent model '%s' for agent '%s' not found; falling back to default model '%s'.",
|
||
agent_model_name,
|
||
agent_id,
|
||
default_model_name,
|
||
)
|
||
|
||
return default_model_name
|
||
|
||
|
||
def _build_model_resolution_config(body: Any, *, thread_id: str) -> dict[str, Any]:
|
||
"""Build the same runtime containers used later by ``start_run``."""
|
||
|
||
config = build_run_config(
|
||
thread_id,
|
||
getattr(body, "config", None),
|
||
getattr(body, "metadata", None),
|
||
assistant_id=getattr(body, "assistant_id", None),
|
||
)
|
||
body_context = getattr(body, "context", None)
|
||
merge_run_context_overrides(
|
||
config,
|
||
body_context if isinstance(body_context, Mapping) else None,
|
||
)
|
||
return config
|
||
|
||
|
||
def _resolve_effective_run_model_name(body: Any) -> str:
|
||
"""Resolve the model that this run will effectively use for concurrency limiting."""
|
||
app_config = get_app_config()
|
||
config = _build_model_resolution_config(body, thread_id="__model_resolution__")
|
||
return _resolve_runtime_model_name(config, app_config)
|
||
|
||
|
||
def _resolve_effective_model_name(
|
||
body: Any,
|
||
app_config: Any | None = None,
|
||
*,
|
||
thread_id: str = "__model_resolution__",
|
||
) -> str:
|
||
"""Compatibility wrapper for resolving the effective run model."""
|
||
if app_config is None:
|
||
return _resolve_effective_run_model_name(body)
|
||
config = _build_model_resolution_config(body, thread_id=thread_id)
|
||
return _resolve_runtime_model_name(config, app_config)
|
||
|
||
|
||
def _resolve_model_concurrency_limit(model_name: str, app_config: Any) -> int | None:
|
||
"""Read legacy per-model max-concurrency values from model config extras."""
|
||
model_config = app_config.get_model_config(model_name)
|
||
extras = getattr(model_config, "model_extra", None) or {}
|
||
for key in ("max_concurrency", "model_concurrency_limit"):
|
||
raw = extras.get(key)
|
||
if raw is None:
|
||
continue
|
||
try:
|
||
limit = int(raw)
|
||
except (TypeError, ValueError):
|
||
continue
|
||
if limit > 0:
|
||
return limit
|
||
return None
|
||
|
||
|
||
def _get_model_concurrency_limit(model_name: str) -> int | None:
|
||
"""Return the configured limit for a model, or ``None`` when limiting is off."""
|
||
settings = load_system_settings().model_concurrency
|
||
if not settings.enabled:
|
||
return None
|
||
|
||
limit = settings.model_limits.get(model_name, settings.default_limit)
|
||
try:
|
||
limit = int(limit)
|
||
except (TypeError, ValueError):
|
||
limit = int(settings.default_limit)
|
||
return max(1, limit)
|
||
|
||
|
||
def _get_user_model_concurrency_limit(model_name: str) -> int | None:
|
||
"""Return the configured per-user limit for a model, or ``None`` when off."""
|
||
settings = load_system_settings().model_concurrency
|
||
if not settings.enabled:
|
||
return None
|
||
|
||
raw_limit = settings.user_model_limits.get(
|
||
model_name,
|
||
getattr(settings, "user_default_limit", 0) or 0,
|
||
)
|
||
try:
|
||
limit = int(raw_limit)
|
||
except (TypeError, ValueError):
|
||
limit = int(getattr(settings, "user_default_limit", 0) or 0)
|
||
if limit < 1:
|
||
return None
|
||
return max(1, limit)
|
||
|
||
|
||
async def _sync_legacy_agent_records(store) -> None:
|
||
"""Import existing folder-backed agents as built-in records.
|
||
|
||
Heavy: ``list_custom_agents()`` scans ``.deer-flow/agents/`` and reads
|
||
``config.yaml`` + ``SOUL.md`` per directory, then ``store.ensure_builtin``
|
||
does one DB round-trip per agent. On a machine with N agent dirs this is
|
||
O(N) file IO + O(N) DB calls — historically the worst per-request hotspot
|
||
in ``start_run`` (observed 30s+ on machines with 100+ residual coordinator
|
||
shells + slow MySQL). **Never call this directly from request handlers** —
|
||
use :func:`_maybe_sync_legacy_agent_records` which applies the same TTL
|
||
gating as ``routers/agents._maybe_sync_legacy_agents``.
|
||
"""
|
||
from deerflow.config.agents_config import list_custom_agents
|
||
|
||
for agent_cfg in list_custom_agents():
|
||
agent_id = agent_cfg.id or agent_cfg.name
|
||
if not agent_id:
|
||
continue
|
||
await store.ensure_builtin(
|
||
{
|
||
"id": agent_id,
|
||
"name": agent_cfg.name,
|
||
"description": agent_cfg.description or "",
|
||
"published": True,
|
||
}
|
||
)
|
||
|
||
|
||
# Throttle per-request legacy-sync. Matches ``routers/agents._LEGACY_SYNC_TTL_SECONDS``
|
||
# semantics — the real sync runs once at startup (lifespan); this is a safety
|
||
# net for environments that bypass the lifespan hook or for freshly dropped
|
||
# agent directories. Lifespan registers itself via the same state by calling
|
||
# the agents router's ``_mark_legacy_sync_fresh()`` after its boot-time sync,
|
||
# so this module's TTL state is independent (a separate gate kept here so we
|
||
# don't import from ``routers/agents`` and create a circular import).
|
||
_LEGACY_SYNC_TTL_SECONDS = 3600.0
|
||
_legacy_sync_state: dict[str, Any] = {"last_at": 0.0, "lock": None}
|
||
|
||
|
||
# Anything slower than this triggers the WARNING-level breakdown in
|
||
# ``start_run``. 500ms gives plenty of headroom for cold caches + a few DB
|
||
# round-trips on healthy systems while still catching the pathological
|
||
# multi-second cases that brought us here.
|
||
_START_RUN_SLOW_THRESHOLD_SECONDS = 0.5
|
||
|
||
|
||
def _mark_legacy_sync_fresh() -> None:
|
||
"""Tell this module's TTL gate that a sync just completed elsewhere.
|
||
|
||
Called by the Gateway lifespan hook after it runs the boot-time legacy
|
||
agent sync (``routers/agents._sync_legacy_agents``). Without this, the
|
||
first request after startup would race past the TTL gate here and trigger
|
||
a redundant rescan even though the data is already fresh.
|
||
"""
|
||
_legacy_sync_state["last_at"] = time.monotonic()
|
||
|
||
|
||
async def _maybe_sync_legacy_agent_records(store) -> None:
|
||
"""TTL-gated wrapper around :func:`_sync_legacy_agent_records`.
|
||
|
||
Steady-state behavior: returns in microseconds inside the TTL window — no
|
||
filesystem walk, no DB round-trips. Outside the TTL window we serialize
|
||
callers under an :class:`asyncio.Lock` so a burst of concurrent
|
||
``/runs/stream`` requests fans into a single rescan instead of N parallel
|
||
fan-outs. Pattern mirrors ``routers/agents._maybe_sync_legacy_agents``.
|
||
"""
|
||
now = time.monotonic()
|
||
if now - float(_legacy_sync_state["last_at"]) < _LEGACY_SYNC_TTL_SECONDS:
|
||
return
|
||
lock = _legacy_sync_state["lock"]
|
||
if lock is None:
|
||
lock = asyncio.Lock()
|
||
_legacy_sync_state["lock"] = lock
|
||
async with lock:
|
||
# Re-check under the lock — a sibling coroutine may have just refreshed.
|
||
if time.monotonic() - float(_legacy_sync_state["last_at"]) < _LEGACY_SYNC_TTL_SECONDS:
|
||
return
|
||
t_sync_start = time.perf_counter()
|
||
await _sync_legacy_agent_records(store)
|
||
elapsed = time.perf_counter() - t_sync_start
|
||
_legacy_sync_state["last_at"] = time.monotonic()
|
||
# When this fires it's a "first request of the hour" warmup — log it
|
||
# so operators can see the spike isn't an isolated bug. Anything
|
||
# over 2s usually means a lot of agent directories + slow IO/DB.
|
||
logger.info(
|
||
"[start-run-timing] legacy-agent sync warmed cache in %.3fs (next sync in %.0fs)",
|
||
elapsed,
|
||
_LEGACY_SYNC_TTL_SECONDS,
|
||
)
|
||
|
||
|
||
async def _enforce_agent_access(body: Any, request: Request) -> None:
|
||
agent_ids = _collect_requested_agent_ids(body)
|
||
if not agent_ids:
|
||
return
|
||
|
||
from app.gateway.deps import get_agent_store
|
||
from deerflow.config.agents_config import validate_agent_id
|
||
from deerflow.runtime.user_context import get_effective_user_id
|
||
|
||
store = get_agent_store(request)
|
||
await _maybe_sync_legacy_agent_records(store)
|
||
user_id = get_effective_user_id()
|
||
# Admins see every agent in the admin console listing (?admin=true), so
|
||
# they may also start a run against any of them.
|
||
# ``request`` may be an in-process virtual request (SimpleNamespace) from
|
||
# the roundtable job gateway / scheduled-task runtime — those carry no
|
||
# ``state`` attribute, so the whole chain must be getattr-guarded
|
||
# (a bare ``request.state`` raises AttributeError and kills the run).
|
||
is_admin = getattr(getattr(getattr(request, "state", None), "user", None), "system_role", None) == "admin"
|
||
for agent_id in agent_ids:
|
||
try:
|
||
normalized = validate_agent_id(agent_id)
|
||
except ValueError as exc:
|
||
raise HTTPException(status_code=422, detail=str(exc)) from exc
|
||
if normalized is None:
|
||
continue
|
||
if await store.get_visible(normalized, user_id) is not None:
|
||
continue
|
||
if is_admin and await store.get_any(normalized) is not None:
|
||
continue
|
||
raise HTTPException(status_code=404, detail=f"Agent {normalized} not found")
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Run lifecycle
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
async def _fail_unstarted_run(
|
||
*,
|
||
run_mgr: RunManager,
|
||
run_ctx: Any,
|
||
record: RunRecord,
|
||
thread_id: str,
|
||
concurrency_gate: Any | None,
|
||
redis_lease_acquired: bool,
|
||
lease_metadata: dict[str, Any],
|
||
exc: BaseException,
|
||
) -> None:
|
||
"""Release a run that failed after registration but before task launch."""
|
||
|
||
error_message = f"{type(exc).__name__}: {exc}"[:1000]
|
||
try:
|
||
# Marking the record terminal removes it from create_or_reject's
|
||
# in-flight set and persists the outcome even though no worker/finally
|
||
# block will run for this pre-launch failure.
|
||
await run_mgr.set_status(record.run_id, RunStatus.error, error=error_message)
|
||
except Exception:
|
||
logger.warning("Failed to close unstarted run %s", record.run_id, exc_info=True)
|
||
try:
|
||
await run_ctx.thread_store.update_status(thread_id, "error")
|
||
except Exception:
|
||
logger.debug("Failed to mark thread %s after run preparation error", sanitize_log_param(thread_id))
|
||
if concurrency_gate is not None and redis_lease_acquired:
|
||
try:
|
||
await concurrency_gate.release(**lease_metadata)
|
||
except Exception:
|
||
logger.warning("Failed to release concurrency lease for unstarted run %s", record.run_id, exc_info=True)
|
||
|
||
|
||
@bounded_worker_run
|
||
async def start_run(
|
||
body: Any,
|
||
thread_id: str,
|
||
request: Request,
|
||
*,
|
||
enforce_access: bool = True,
|
||
run_scope: str | None = None,
|
||
) -> RunRecord:
|
||
"""Create a RunRecord and launch the background agent task.
|
||
|
||
Parameters
|
||
----------
|
||
body : RunCreateRequest
|
||
The validated request body (typed as Any to avoid circular import
|
||
with the router module that defines the Pydantic model).
|
||
thread_id : str
|
||
Target thread.
|
||
request : Request
|
||
FastAPI request — used to retrieve singletons from ``app.state``.
|
||
|
||
Timing observability
|
||
--------------------
|
||
Each phase records ``perf_counter`` deltas. When the total wallclock from
|
||
function entry to ``run_agent`` task creation exceeds
|
||
``_START_RUN_SLOW_THRESHOLD_SECONDS`` we emit a ``[start-run-timing]``
|
||
breakdown at WARNING level. Stays at DEBUG when fast, so steady-state
|
||
logs are not polluted. Investigate spikes by grep'ing
|
||
``[start-run-timing]`` together with ``[upstream-timing]`` (multi-agent
|
||
router) and ``[leader-timing] / [special-timing]`` (圆桌 leader/special).
|
||
"""
|
||
t_entry = time.perf_counter()
|
||
from app.gateway.analysis_projects import prepare_project_run
|
||
from app.gateway.knowledge_research import (
|
||
inject_validated_knowledge_research_context,
|
||
prepare_knowledge_research_run,
|
||
)
|
||
|
||
body, analysis_project = await prepare_project_run(body, thread_id, request)
|
||
body, knowledge_research = await prepare_knowledge_research_run(
|
||
body,
|
||
thread_id,
|
||
request,
|
||
)
|
||
bridge = get_stream_bridge(request)
|
||
run_mgr = get_run_manager(request)
|
||
run_ctx = get_run_context(request)
|
||
runtime_app_config = getattr(run_ctx, "app_config", None) or get_app_config()
|
||
concurrency_gate = get_concurrency_gate(request)
|
||
|
||
disconnect = DisconnectMode.cancel if body.on_disconnect == "cancel" else DisconnectMode.continue_
|
||
|
||
t_pre_enforce = time.perf_counter()
|
||
# ``enforce_access=False`` is for intentionally open/unauthenticated entry
|
||
# points (e.g. /api/open/3qfx/ask) that resolve the agent themselves from an
|
||
# admin-configured mapping rather than from a per-user-visible assistant_id.
|
||
# Those run under the "default" bucket, where a user-owned unpublished agent
|
||
# would otherwise 404 the access check.
|
||
if enforce_access:
|
||
await _enforce_agent_access(body, request)
|
||
t_post_enforce = time.perf_counter()
|
||
|
||
effective_user_id = get_effective_user_id()
|
||
effective_run_scope = _resolve_run_scope(body, request, run_scope)
|
||
graph_input = normalize_input(body.input)
|
||
requested_context = getattr(body, "context", None)
|
||
requested_retrieval_mode = (
|
||
str(requested_context.get("knowledge_retrieval_mode") or "")
|
||
.strip()
|
||
.lower()
|
||
if isinstance(requested_context, Mapping)
|
||
else ""
|
||
)
|
||
fixed_answer = (
|
||
None
|
||
if analysis_project or knowledge_research is not None or requested_retrieval_mode == "smart_reasoning"
|
||
else await _find_fixed_answer(
|
||
body=body,
|
||
request=request,
|
||
graph_input=graph_input,
|
||
run_scope=effective_run_scope,
|
||
)
|
||
)
|
||
if fixed_answer is not None:
|
||
effective_model_name = "fixed-answer"
|
||
model_concurrency_limit = None
|
||
user_model_concurrency_limit = None
|
||
else:
|
||
effective_model_name = _resolve_effective_model_name(
|
||
body,
|
||
runtime_app_config,
|
||
thread_id=thread_id,
|
||
)
|
||
model_concurrency_limit = _get_model_concurrency_limit(effective_model_name)
|
||
user_model_concurrency_limit = _get_user_model_concurrency_limit(effective_model_name) if effective_run_scope == RUN_SCOPE_EXTERNAL_ROOT else None
|
||
run_id = str(uuid.uuid4())
|
||
|
||
run_metadata = dict(body.metadata or {})
|
||
run_metadata["model_name"] = effective_model_name
|
||
run_metadata["user_id"] = effective_user_id
|
||
run_metadata["run_scope"] = effective_run_scope
|
||
if requested_retrieval_mode == "smart_reasoning":
|
||
# Avoid a transient reference-mode frame before the database-backed
|
||
# smart-mode gate and knowledge scope finish validation below.
|
||
run_metadata.update({"rag_mode_enabled": False, "rag_mode": False})
|
||
if fixed_answer is not None:
|
||
run_metadata["fixed_answer"] = True
|
||
run_metadata["fixed_question_id"] = fixed_answer["id"]
|
||
|
||
redis_lease_acquired = False
|
||
redis_gate_degraded = True
|
||
user_model_accounted = user_model_concurrency_limit is not None
|
||
lease_metadata = {
|
||
"run_id": run_id,
|
||
"thread_id": thread_id,
|
||
"model_name": effective_model_name,
|
||
"user_id": effective_user_id,
|
||
"user_model_accounted": user_model_accounted,
|
||
}
|
||
if concurrency_gate is not None and fixed_answer is None:
|
||
acquire = await concurrency_gate.try_acquire(
|
||
run_id=run_id,
|
||
thread_id=thread_id,
|
||
model_name=effective_model_name,
|
||
user_id=effective_user_id,
|
||
assistant_id=body.assistant_id or "",
|
||
model_limit=model_concurrency_limit,
|
||
user_model_limit=user_model_concurrency_limit,
|
||
reject_thread=body.multitask_strategy == "reject",
|
||
)
|
||
redis_lease_acquired = acquire.acquired
|
||
redis_gate_degraded = acquire.degraded
|
||
if acquire.rejected:
|
||
if acquire.reason == "thread":
|
||
raise HTTPException(
|
||
status_code=409,
|
||
detail="Current thread already has an active run. Please wait or stop it before sending another message.",
|
||
)
|
||
if acquire.reason == "user_model":
|
||
raise HTTPException(
|
||
status_code=429,
|
||
detail="This account has reached the concurrency limit for the selected model. Please try again later.",
|
||
)
|
||
raise HTTPException(status_code=429, detail="The selected model is busy. Please try again later.")
|
||
|
||
try:
|
||
record = await run_mgr.create_or_reject(
|
||
thread_id,
|
||
body.assistant_id,
|
||
on_disconnect=disconnect,
|
||
metadata=run_metadata,
|
||
kwargs={"input": body.input, "config": body.config},
|
||
multitask_strategy=body.multitask_strategy,
|
||
model_concurrency_limit=(effective_model_name, model_concurrency_limit) if redis_gate_degraded and model_concurrency_limit is not None else None,
|
||
user_model_concurrency_limit=(effective_model_name, effective_user_id, user_model_concurrency_limit) if redis_gate_degraded and user_model_concurrency_limit is not None else None,
|
||
run_id=run_id,
|
||
)
|
||
except ModelConcurrencyLimitError as exc:
|
||
if concurrency_gate is not None and redis_lease_acquired:
|
||
await concurrency_gate.release(**lease_metadata)
|
||
if getattr(exc, "scope", "") == "user_model":
|
||
detail = "This account has reached the concurrency limit for the selected model. Please try again later."
|
||
else:
|
||
detail = "The selected model is busy. Please try again later."
|
||
raise HTTPException(status_code=429, detail=detail) from exc
|
||
except ConflictError as exc:
|
||
if concurrency_gate is not None and redis_lease_acquired:
|
||
await concurrency_gate.release(**lease_metadata)
|
||
# 同一会话上一轮回答还没结束，又提交了新问题 → 拒绝。给用户中文提示，
|
||
# 而不是把内部英文「Thread ... already has an active run」直接抛到前端。
|
||
raise HTTPException(
|
||
status_code=409,
|
||
detail="Current thread already has an active run. Please wait or stop it before sending another message.",
|
||
) from exc
|
||
except UnsupportedStrategyError as exc:
|
||
if concurrency_gate is not None and redis_lease_acquired:
|
||
await concurrency_gate.release(**lease_metadata)
|
||
raise HTTPException(status_code=501, detail=str(exc)) from exc
|
||
except Exception:
|
||
if concurrency_gate is not None and redis_lease_acquired:
|
||
await concurrency_gate.release(**lease_metadata)
|
||
raise
|
||
t_post_create = time.perf_counter()
|
||
|
||
# Upsert thread metadata so the thread appears in /threads/search,
|
||
# even for threads that were never explicitly created via POST /threads
|
||
# (e.g. stateless runs).
|
||
requested_rag_mode = _rag_mode_requested(body)
|
||
effective_rag_mode = requested_rag_mode
|
||
existing_thread_metadata: dict[str, Any] = {}
|
||
try:
|
||
existing = await run_ctx.thread_store.get(thread_id)
|
||
if existing is not None and isinstance(existing.get("metadata"), Mapping):
|
||
existing_thread_metadata = dict(existing["metadata"])
|
||
effective_rag_mode = requested_rag_mode or _thread_rag_mode_enabled(existing)
|
||
if existing is None:
|
||
metadata = dict(body.metadata or {})
|
||
if requested_retrieval_mode == "smart_reasoning":
|
||
metadata.update({"rag_mode_enabled": False, "rag_mode": False})
|
||
elif effective_rag_mode:
|
||
metadata["rag_mode_enabled"] = True
|
||
metadata["rag_mode"] = True
|
||
await run_ctx.thread_store.create(
|
||
thread_id,
|
||
assistant_id=body.assistant_id,
|
||
metadata=metadata,
|
||
)
|
||
else:
|
||
await run_ctx.thread_store.update_status(thread_id, "running")
|
||
if requested_rag_mode and not _thread_rag_mode_enabled(existing):
|
||
await run_ctx.thread_store.update_metadata(
|
||
thread_id,
|
||
{"rag_mode_enabled": True, "rag_mode": True},
|
||
)
|
||
effective_rag_mode = True
|
||
# SDK 的 ``client.threads.create()`` 不传 assistant_id —— 先建出来空的 thread，
|
||
# 真正提交 run 时才知道用哪个 assistant。这里回填一次，避免历史列表 / 按
|
||
# assistant 筛选时拿不到关联。
|
||
if not existing.get("assistant_id") and body.assistant_id:
|
||
await run_ctx.thread_store.update_assistant_id(thread_id, body.assistant_id)
|
||
except Exception:
|
||
logger.warning("Failed to upsert thread_meta for %s (non-fatal)", sanitize_log_param(thread_id))
|
||
t_post_thread_upsert = time.perf_counter()
|
||
|
||
try:
|
||
if fixed_answer is not None:
|
||
from deerflow.agents.fixed_answer_agent import make_fixed_answer_agent
|
||
|
||
def agent_factory(config): # noqa: ANN001, ARG001 - worker factory contract
|
||
return make_fixed_answer_agent(
|
||
question=str(fixed_answer["question"]),
|
||
answer=str(fixed_answer["answer"]),
|
||
fixed_question_id=str(fixed_answer["id"]),
|
||
tokens_per_second=int(fixed_answer.get("tokens_per_second", 100)),
|
||
)
|
||
else:
|
||
agent_factory = resolve_agent_factory(body.assistant_id)
|
||
except Exception as exc:
|
||
await _fail_unstarted_run(
|
||
run_mgr=run_mgr,
|
||
run_ctx=run_ctx,
|
||
record=record,
|
||
thread_id=thread_id,
|
||
concurrency_gate=concurrency_gate,
|
||
redis_lease_acquired=redis_lease_acquired,
|
||
lease_metadata=lease_metadata,
|
||
exc=exc,
|
||
)
|
||
raise
|
||
|
||
# Admin-configurable prompt prefix: when the caller opted in via
|
||
# context.prompt_prefix_enabled, tag the human message with the configured
|
||
# prefix (stored in additional_kwargs, not concatenated into content, so the
|
||
# user never sees it — PromptPrefixMiddleware injects it for the LLM only).
|
||
# The frontend only sends the flag on the opening turn, so this naturally
|
||
# applies once. We intentionally do NOT gate on thread-row existence: the
|
||
# thread_meta row is frequently pre-created, which previously skipped the
|
||
# prefix for genuine first turns.
|
||
try:
|
||
knowledge_retrieval_mode = await resolve_knowledge_retrieval_mode(
|
||
request,
|
||
body,
|
||
)
|
||
if fixed_answer is None and _prompt_prefix_opted_in(body):
|
||
prefix_settings = load_system_settings().prompt_prefix
|
||
if prefix_settings.enabled and prefix_settings.prompt_prefix.strip():
|
||
graph_input = _prepend_prompt_prefix(graph_input, prefix_settings.prompt_prefix.strip())
|
||
llmwiki_rag = (
|
||
LlmWikiRagContext([], [], 0, "", [], "")
|
||
if fixed_answer is not None
|
||
else await (
|
||
build_llmwiki_smart_reasoning_context
|
||
if knowledge_retrieval_mode == "smart_reasoning"
|
||
else build_llmwiki_rag_context
|
||
)(
|
||
request=request,
|
||
body=body,
|
||
graph_input=graph_input,
|
||
fallback_metadata=existing_thread_metadata,
|
||
)
|
||
)
|
||
except Exception as exc:
|
||
await _fail_unstarted_run(
|
||
run_mgr=run_mgr,
|
||
run_ctx=run_ctx,
|
||
record=record,
|
||
thread_id=thread_id,
|
||
concurrency_gate=concurrency_gate,
|
||
redis_lease_acquired=redis_lease_acquired,
|
||
lease_metadata=lease_metadata,
|
||
exc=exc,
|
||
)
|
||
raise
|
||
effective_rag_mode = _effective_rag_mode_after_llmwiki(
|
||
body=body,
|
||
rag=llmwiki_rag,
|
||
inherited=effective_rag_mode,
|
||
)
|
||
if knowledge_retrieval_mode == "smart_reasoning":
|
||
# Smart retrieval is a distinct evidence-gathering mode, not classic
|
||
# RAG/reference display mode. A stale thread flag must not leak into
|
||
# prompt injection or the runtime context.
|
||
effective_rag_mode = False
|
||
has_llmwiki_observability = bool(llmwiki_rag.prompt or llmwiki_rag.sources or llmwiki_rag.trace)
|
||
knowledge_scope_active = bool(llmwiki_rag.knowledge_base_ids or llmwiki_rag.assistant_knowledge_base_ids)
|
||
run_metadata["knowledge_retrieval_mode"] = knowledge_retrieval_mode
|
||
record.metadata["knowledge_retrieval_mode"] = knowledge_retrieval_mode
|
||
if has_llmwiki_observability:
|
||
graph_input = attach_llmwiki_rag_to_graph_input(graph_input, llmwiki_rag)
|
||
run_metadata["llmwiki_knowledge_base_ids"] = llmwiki_rag.knowledge_base_ids
|
||
run_metadata["assistant_knowledge_base_ids"] = llmwiki_rag.assistant_knowledge_base_ids
|
||
run_metadata["llmwiki_rag_source_count"] = llmwiki_rag.source_count
|
||
if llmwiki_rag.knowledge_base_context:
|
||
# Keep the server-resolved identity beside the selected ids. A
|
||
# later turn still re-authorizes the ids and rebuilds this value;
|
||
# the durable copy exists so thread consumers do not have to infer
|
||
# what phrases such as “这个库” referred to.
|
||
run_metadata["llmwiki_knowledge_base_context"] = (
|
||
llmwiki_rag.knowledge_base_context
|
||
)
|
||
record.metadata["llmwiki_knowledge_base_context"] = (
|
||
llmwiki_rag.knowledge_base_context
|
||
)
|
||
selection_metadata_update: dict[str, Any] = {}
|
||
should_persist_retrieval_mode = (
|
||
knowledge_retrieval_mode == "smart_reasoning"
|
||
or "knowledge_retrieval_mode" in existing_thread_metadata
|
||
or has_llmwiki_observability
|
||
or llmwiki_selection_declared(body)
|
||
)
|
||
if should_persist_retrieval_mode:
|
||
selection_metadata_update["knowledge_retrieval_mode"] = knowledge_retrieval_mode
|
||
if has_llmwiki_observability or llmwiki_selection_declared(
|
||
body,
|
||
"llmwiki_knowledge_base_ids",
|
||
):
|
||
selection_metadata_update["llmwiki_knowledge_base_ids"] = llmwiki_rag.knowledge_base_ids
|
||
if has_llmwiki_observability or llmwiki_selection_declared(
|
||
body,
|
||
"assistant_knowledge_base_ids",
|
||
):
|
||
selection_metadata_update["assistant_knowledge_base_ids"] = llmwiki_rag.assistant_knowledge_base_ids
|
||
if (
|
||
llmwiki_rag.knowledge_base_context
|
||
or "llmwiki_knowledge_base_context" in existing_thread_metadata
|
||
):
|
||
selection_metadata_update["llmwiki_knowledge_base_context"] = (
|
||
llmwiki_rag.knowledge_base_context
|
||
)
|
||
_sync_retrieval_rag_metadata(
|
||
run_metadata=run_metadata,
|
||
record_metadata=record.metadata,
|
||
thread_metadata_update=selection_metadata_update,
|
||
knowledge_retrieval_mode=knowledge_retrieval_mode,
|
||
knowledge_scope_active=knowledge_scope_active,
|
||
selection_declared=llmwiki_selection_declared(body),
|
||
explicit_rag_requested=_explicit_rag_mode_requested(body),
|
||
)
|
||
if selection_metadata_update:
|
||
try:
|
||
await run_ctx.thread_store.update_metadata(thread_id, selection_metadata_update)
|
||
except Exception as exc:
|
||
# Retrieval mode and scope are part of the durable run contract.
|
||
# Starting the graph after this write fails can make a later turn
|
||
# silently use a different retrieval policy, so fail the registered
|
||
# but not-yet-started run and release all of its leases instead.
|
||
logger.exception(
|
||
"Failed to persist LLMWiki retrieval metadata for %s",
|
||
sanitize_log_param(thread_id),
|
||
)
|
||
await _fail_unstarted_run(
|
||
run_mgr=run_mgr,
|
||
run_ctx=run_ctx,
|
||
record=record,
|
||
thread_id=thread_id,
|
||
concurrency_gate=concurrency_gate,
|
||
redis_lease_acquired=redis_lease_acquired,
|
||
lease_metadata=lease_metadata,
|
||
exc=exc,
|
||
)
|
||
raise
|
||
if llmwiki_rag.trace:
|
||
# Keep the exact pre-retrieval trace on the live run so the initial
|
||
# SSE metadata frame is inspectable in browser DevTools before the
|
||
# answer finishes. The attached HumanMessage remains the durable
|
||
# copy used by /reference-batches after completion.
|
||
run_metadata["knowledge_retrieval_trace"] = llmwiki_rag.trace
|
||
record.metadata["knowledge_retrieval_trace"] = llmwiki_rag.trace
|
||
graph_input = _apply_classic_rag_citation_prompt(
|
||
graph_input,
|
||
enabled=effective_rag_mode,
|
||
knowledge_retrieval_mode=knowledge_retrieval_mode,
|
||
)
|
||
try:
|
||
config = build_run_config(thread_id, body.config, body.metadata, assistant_id=body.assistant_id)
|
||
except Exception as exc:
|
||
await _fail_unstarted_run(
|
||
run_mgr=run_mgr,
|
||
run_ctx=run_ctx,
|
||
record=record,
|
||
thread_id=thread_id,
|
||
concurrency_gate=concurrency_gate,
|
||
redis_lease_acquired=redis_lease_acquired,
|
||
lease_metadata=lease_metadata,
|
||
exc=exc,
|
||
)
|
||
raise
|
||
|
||
# Merge DeerFlow-specific context overrides into both ``configurable`` and ``context``.
|
||
# The ``context`` field is a custom extension for the langgraph-compat layer
|
||
# that carries agent configuration (model_name, thinking_enabled, etc.).
|
||
# Only agent-relevant keys are forwarded; unknown keys (e.g. thread_id) are ignored.
|
||
body_context = dict(getattr(body, "context", None) or {})
|
||
body_context["knowledge_retrieval_mode"] = knowledge_retrieval_mode
|
||
if effective_rag_mode and knowledge_retrieval_mode != "smart_reasoning":
|
||
body_context["rag_mode_enabled"] = True
|
||
elif knowledge_retrieval_mode == "smart_reasoning":
|
||
body_context["rag_mode_enabled"] = False
|
||
if "llmwiki_knowledge_base_ids" not in body_context and llmwiki_rag.knowledge_base_ids:
|
||
body_context["llmwiki_knowledge_base_ids"] = llmwiki_rag.knowledge_base_ids
|
||
if "assistant_knowledge_base_ids" not in body_context and llmwiki_rag.assistant_knowledge_base_ids:
|
||
body_context["assistant_knowledge_base_ids"] = llmwiki_rag.assistant_knowledge_base_ids
|
||
# Promote top-level exclusions before deriving the strict scope so caller
|
||
# policy is retained instead of being replaced by the knowledge policy.
|
||
for top_level_key in ("excluded_tools", "skill_stop_names", "skill_stop_message"):
|
||
value = getattr(body, top_level_key, None)
|
||
if value is not None and top_level_key not in body_context:
|
||
body_context[top_level_key] = value
|
||
if knowledge_retrieval_mode == "smart_reasoning":
|
||
# Smart retrieval is an optional tool surface in the normal agent. It
|
||
# retains the validated KB ids but deliberately does not activate the
|
||
# classic fail-closed tool/skill/MCP isolation policy.
|
||
body_context["knowledge_space_strict"] = False
|
||
strict_scope_active = False
|
||
else:
|
||
strict_scope_active = apply_knowledge_space_strict_scope(
|
||
body_context,
|
||
selection_requested=llmwiki_selection_requested(body),
|
||
knowledge_base_ids=llmwiki_rag.knowledge_base_ids,
|
||
assistant_knowledge_base_ids=llmwiki_rag.assistant_knowledge_base_ids,
|
||
app_config=getattr(request.app.state, "config", None),
|
||
)
|
||
if strict_scope_active:
|
||
run_metadata["knowledge_space_strict"] = True
|
||
elif not llmwiki_rag.knowledge_base_ids:
|
||
raw_excluded_tools = body_context.get("excluded_tools")
|
||
if isinstance(raw_excluded_tools, str):
|
||
excluded_tools = [raw_excluded_tools] if raw_excluded_tools.strip() else []
|
||
elif isinstance(raw_excluded_tools, (list, tuple, set)):
|
||
excluded_tools = [str(item) for item in raw_excluded_tools if str(item).strip()]
|
||
else:
|
||
excluded_tools = []
|
||
unavailable_classic_tools = {
|
||
"llmwiki_browse_wiki_index",
|
||
"llmwiki_read_wiki_page",
|
||
}
|
||
if not llmwiki_rag.assistant_knowledge_base_ids:
|
||
unavailable_classic_tools.add("llmwiki_search")
|
||
missing_classic_tools = [
|
||
name for name in sorted(unavailable_classic_tools) if name not in excluded_tools
|
||
]
|
||
if missing_classic_tools:
|
||
body_context["excluded_tools"] = [*excluded_tools, *missing_classic_tools]
|
||
merge_run_context_overrides(config, body_context or None)
|
||
# This identity summary is generated only after server-side authorization.
|
||
# Put it in both LangGraph scopes so model/middleware code can resolve
|
||
# natural references to the selected library without trusting client
|
||
# metadata or inspecting a sandbox mount.
|
||
for runtime_scope in ("configurable", "context"):
|
||
runtime_target = config.setdefault(runtime_scope, {})
|
||
if isinstance(runtime_target, dict):
|
||
runtime_target["llmwiki_knowledge_base_context"] = list(
|
||
llmwiki_rag.knowledge_base_context
|
||
)
|
||
enforce_knowledge_retrieval_mode_run_config(config, knowledge_retrieval_mode)
|
||
if fixed_answer is not None:
|
||
reasoning_effort_capability = False
|
||
effective_model_supports_thinking = False
|
||
else:
|
||
# Resolve again from the actual final RunnableConfig. This is the
|
||
# security boundary: pre-seeded config.context values win exactly as
|
||
# they do in lead_agent._get_runtime_config, so capability A can never
|
||
# authorize a different runtime model B.
|
||
final_model_name = _resolve_runtime_model_name(config, runtime_app_config)
|
||
if final_model_name != effective_model_name:
|
||
logger.warning(
|
||
"Final runtime model '%s' differed from early concurrency model '%s'; applying final model policy.",
|
||
sanitize_log_param(final_model_name),
|
||
sanitize_log_param(effective_model_name),
|
||
)
|
||
effective_model_name = final_model_name
|
||
run_metadata["model_name"] = final_model_name
|
||
record.metadata["model_name"] = final_model_name
|
||
effective_model_config = runtime_app_config.get_model_config(final_model_name)
|
||
effective_model_supports_thinking = bool(
|
||
effective_model_config
|
||
and getattr(effective_model_config, "supports_thinking", False)
|
||
)
|
||
reasoning_effort_capability, default_reasoning_effort = await _reasoning_effort_policy(
|
||
request,
|
||
final_model_name,
|
||
)
|
||
try:
|
||
normalized_reasoning_effort = enforce_reasoning_effort_run_config(
|
||
config,
|
||
capability_enabled=reasoning_effort_capability,
|
||
model_supports_thinking=effective_model_supports_thinking,
|
||
default_reasoning_effort=default_reasoning_effort,
|
||
)
|
||
except Exception as exc:
|
||
await _fail_unstarted_run(
|
||
run_mgr=run_mgr,
|
||
run_ctx=run_ctx,
|
||
record=record,
|
||
thread_id=thread_id,
|
||
concurrency_gate=concurrency_gate,
|
||
redis_lease_acquired=redis_lease_acquired,
|
||
lease_metadata=lease_metadata,
|
||
exc=exc,
|
||
)
|
||
raise
|
||
if normalized_reasoning_effort is not None:
|
||
run_metadata["reasoning_effort"] = normalized_reasoning_effort
|
||
record.metadata["reasoning_effort"] = normalized_reasoning_effort
|
||
else:
|
||
run_metadata.pop("reasoning_effort", None)
|
||
record.metadata.pop("reasoning_effort", None)
|
||
if strict_scope_active:
|
||
enforce_knowledge_space_strict_run_config(
|
||
config,
|
||
knowledge_base_ids=llmwiki_rag.knowledge_base_ids,
|
||
assistant_knowledge_base_ids=llmwiki_rag.assistant_knowledge_base_ids,
|
||
excluded_tools=list(body_context.get("excluded_tools") or []),
|
||
app_config=getattr(request.app.state, "config", None),
|
||
)
|
||
# Strict-scope merging intentionally preserves caller exclusions. In
|
||
# smart mode the five validated retrieval tools are mandatory, so make
|
||
# the mode policy authoritative once more after that merge.
|
||
enforce_knowledge_retrieval_mode_run_config(
|
||
config,
|
||
knowledge_retrieval_mode,
|
||
)
|
||
if analysis_project:
|
||
# These keys are deliberately absent from the public context whitelist.
|
||
# Only the validated project binding/checkpoint can supply them.
|
||
inject_validated_analysis_project_context(config, body.context)
|
||
if knowledge_research:
|
||
inject_validated_knowledge_research_context(config, body.context)
|
||
|
||
stream_modes = _with_knowledge_retrieval_progress_stream(
|
||
normalize_stream_modes(body.stream_mode),
|
||
knowledge_retrieval_mode,
|
||
)
|
||
|
||
t_pre_task = time.perf_counter()
|
||
heartbeat_interval_seconds = 20
|
||
if concurrency_gate is not None:
|
||
heartbeat_interval_seconds = int(getattr(getattr(concurrency_gate, "config", None), "heartbeat_interval_seconds", 20) or 20)
|
||
try:
|
||
after_success = None
|
||
if analysis_project and analysis_project.get("mode") == "qa":
|
||
from app.gateway.analysis_project_memory import enqueue_completed_project_run
|
||
|
||
async def _enqueue_project_memory() -> None:
|
||
await enqueue_completed_project_run(
|
||
app=request.app,
|
||
project=analysis_project,
|
||
thread_id=thread_id,
|
||
run_id=run_id,
|
||
)
|
||
|
||
after_success = _enqueue_project_memory
|
||
task = asyncio.create_task(
|
||
_run_agent_with_concurrency_lease(
|
||
concurrency_gate=concurrency_gate,
|
||
lease_acquired=redis_lease_acquired,
|
||
lease_metadata=lease_metadata,
|
||
heartbeat_interval_seconds=heartbeat_interval_seconds,
|
||
after_success=after_success,
|
||
run_agent_kwargs={
|
||
"bridge": bridge,
|
||
"run_manager": run_mgr,
|
||
"record": record,
|
||
"ctx": run_ctx,
|
||
"agent_factory": agent_factory,
|
||
"graph_input": graph_input,
|
||
"config": config,
|
||
"stream_modes": stream_modes,
|
||
"stream_subgraphs": body.stream_subgraphs,
|
||
"interrupt_before": body.interrupt_before,
|
||
"interrupt_after": body.interrupt_after,
|
||
"command": getattr(body, "command", None),
|
||
},
|
||
)
|
||
)
|
||
except Exception as exc:
|
||
await _fail_unstarted_run(
|
||
run_mgr=run_mgr,
|
||
run_ctx=run_ctx,
|
||
record=record,
|
||
thread_id=thread_id,
|
||
concurrency_gate=concurrency_gate,
|
||
redis_lease_acquired=redis_lease_acquired,
|
||
lease_metadata=lease_metadata,
|
||
exc=exc,
|
||
)
|
||
raise
|
||
record.task = task
|
||
run_mgr.watch_completion(record)
|
||
|
||
# Title sync is handled by worker.py's finally block which reads the
|
||
# title from the checkpoint and calls thread_store.update_display_name
|
||
# after the run completes.
|
||
|
||
total = t_pre_task - t_entry
|
||
if total >= _START_RUN_SLOW_THRESHOLD_SECONDS:
|
||
# Slow path: emit a phase breakdown so on-call can identify the
|
||
# culprit without re-running with extra instrumentation. The most
|
||
# common cause is a stale ``_maybe_sync_legacy_agent_records`` TTL
|
||
# window firing on a machine with hundreds of agent directories.
|
||
logger.warning(
|
||
"[start-run-timing] thread=%s total=%.3fs enforce_access=%.3fs run_create=%.3fs thread_upsert=%.3fs build_payload=%.3fs",
|
||
sanitize_log_param(thread_id),
|
||
total,
|
||
t_post_enforce - t_pre_enforce,
|
||
t_post_create - t_post_enforce,
|
||
t_post_thread_upsert - t_post_create,
|
||
t_pre_task - t_post_thread_upsert,
|
||
)
|
||
else:
|
||
logger.debug(
|
||
"[start-run-timing] thread=%s total=%.3fs enforce=%.3fs create=%.3fs upsert=%.3fs build=%.3fs",
|
||
sanitize_log_param(thread_id),
|
||
total,
|
||
t_post_enforce - t_pre_enforce,
|
||
t_post_create - t_post_enforce,
|
||
t_post_thread_upsert - t_post_create,
|
||
t_pre_task - t_post_thread_upsert,
|
||
)
|
||
|
||
return record
|
||
|
||
|
||
async def sse_consumer(
|
||
bridge: StreamBridge,
|
||
record: RunRecord,
|
||
request: Request,
|
||
run_mgr: RunManager,
|
||
):
|
||
"""Async generator that yields SSE frames from the bridge.
|
||
|
||
The ``finally`` block implements ``on_disconnect`` semantics:
|
||
- ``cancel``: abort the background task on client disconnect.
|
||
- ``continue``: let the task run so the client can ``joinStream``.
|
||
|
||
``is_disconnected()`` is only consulted in cancel mode. In continue
|
||
(resumable) mode a false-positive disconnect would *cleanly* end the
|
||
HTTP stream; the LangGraph SDK treats that as success, clears
|
||
``lg:stream:{threadId}``, and flips ``isLoading`` off — while the
|
||
backend run keeps the thread occupied, so the next send 409s.
|
||
A truly gone client fails on the next ``yield`` instead.
|
||
"""
|
||
last_event_id = request.headers.get("Last-Event-ID")
|
||
# Completed records outlive the short SSE replay buffer. Rejoining after
|
||
# buffer retirement must end cleanly, not create an empty endless stream.
|
||
has_stream = getattr(bridge, "has_stream", None)
|
||
if record.status not in (RunStatus.pending, RunStatus.running) and callable(has_stream) and not has_stream(record.run_id):
|
||
yield format_sse("end", None)
|
||
return
|
||
try:
|
||
async for entry in bridge.subscribe(record.run_id, last_event_id=last_event_id):
|
||
if record.on_disconnect == DisconnectMode.cancel and await request.is_disconnected():
|
||
break
|
||
|
||
if entry is HEARTBEAT_SENTINEL:
|
||
yield ": heartbeat\n\n"
|
||
continue
|
||
|
||
if entry is END_SENTINEL:
|
||
yield format_sse("end", None, event_id=entry.id or None)
|
||
return
|
||
|
||
yield format_sse(entry.event, entry.data, event_id=entry.id or None)
|
||
|
||
except (ConnectionError, BrokenPipeError, ConnectionResetError, ConnectionAbortedError):
|
||
logger.info("SSE client connection lost run_id=%s thread_id=%s", record.run_id, record.thread_id)
|
||
|
||
finally:
|
||
if record.status in (RunStatus.pending, RunStatus.running):
|
||
if record.on_disconnect == DisconnectMode.cancel:
|
||
await run_mgr.cancel(record.run_id)
|