Merge branch 'ji' into master

This commit is contained in:
Administrator committed 2026-06-12 18:28:13 +08:00
commit e74ef4b7dc
81 files changed
+3712 -1625

No files matched your search

@@ -100,7 +100,7 @@ async def _ensure_admin_user(app: FastAPI) -> None:
After admin creation, migrate orphan threads from the LangGraph
store (metadata.user_id unset) to the admin account. This is the
"no-auth → with-auth" upgrade path: users who ran DeerFlow without
"no-auth ? with-auth" upgrade path: users who ran DeerFlow without
authentication have existing LangGraph thread data that needs an
owner assigned.
First boot (no admin exists):
@@ -108,7 +108,7 @@ async def _ensure_admin_user(app: FastAPI) -> None:
- The operator must visit ``/setup`` to create the first admin.
Subsequent boots (admin already exists):
- Runs the one-time "no-auth → with-auth" orphan thread migration for
- Runs the one-time "no-auth ? with-auth" orphan thread migration for
existing LangGraph thread metadata that has no owner_id.
No SQL persistence migration is needed: the four user_id columns
@@ -138,12 +138,12 @@ async def _ensure_admin_user(app: FastAPI) -> None:
if admin_count == 0:
logger.info("=" * 60)
logger.info(" First boot detected — no admin account exists.")
logger.info(" First boot detected ? no admin account exists.")
logger.info(" Visit /setup to complete admin account creation.")
logger.info("=" * 60)
return
# Admin already exists — run orphan thread migration for any
# Admin already exists ? run orphan thread migration for any
# LangGraph thread metadata that pre-dates the auth module.
async with sf() as session:
stmt = select(UserRow).where(UserRow.system_role == "admin").limit(1)
@@ -154,8 +154,8 @@ async def _ensure_admin_user(app: FastAPI) -> None:
admin_id = str(row.id)
# LangGraph store orphan migration — non-fatal.
# This covers the "no-auth → with-auth" upgrade path for users
# LangGraph store orphan migration ? non-fatal.
# This covers the "no-auth ? with-auth" upgrade path for users
# whose existing LangGraph thread metadata has no user_id set.
store = getattr(app.state, "store", None)
if store is not None:
@@ -217,7 +217,7 @@ async def _sync_legacy_skills(app: FastAPI) -> None:
stmt = select(UserRow).where(UserRow.system_role == "admin").limit(1)
row = (await session.execute(stmt)).scalar_one_or_none()
if row is None:
# No admin yet — leave ownership null; first admin is responsible for
# No admin yet ? leave ownership null; first admin is responsible for
# claiming via the UI once they exist.
admin_id: str | None = None
else:
@@ -278,7 +278,7 @@ async def _run_db_migrations() -> None:
engine = get_engine()
if engine is None:
logger.info("Persistence backend=memory — skipping Alembic migrations")
logger.info("Persistence backend=memory ? skipping Alembic migrations")
return
if os.getenv("DEER_FLOW_SKIP_ALEMBIC", "").lower() in {"1", "true", "yes"}:
@@ -327,28 +327,28 @@ async def _run_db_migrations() -> None:
raise result
logger.info("Database migrations applied (head)")
except Exception:
logger.exception("Database migration failed — schema may be out of date")
logger.exception("Database migration failed ? schema may be out of date")
# Skill curator 调度参数。
# - 检查间隔:每隔这么久检查一次"需不需要进化";真正是否执行由 should_run_now() 判定。
# - 启动延迟:启动后先等这么久再首次检查,避开启动高峰。
_CURATOR_CHECK_INTERVAL_SECONDS = 3600 # 1 小时
_CURATOR_STARTUP_DELAY_SECONDS = 300 # 5 分钟
# Skill curator ?????
# - ????:?????????"??????";??????? should_run_now() ???
# - ????:?????????????,???????
_CURATOR_CHECK_INTERVAL_SECONDS = 3600 # 1 ??
_CURATOR_STARTUP_DELAY_SECONDS = 300 # 5 ??
def _acquire_singleton_lock(lock_path):
"""跨平台非阻塞独占文件锁,用于多 worker 下的 leader 选举。
"""???????????,??? worker ?? leader ???
成功 -> 返回持有锁的文件句柄(调用方须保持其存活,直到进程退出/主动释放);
抢不到锁(已有其它 worker 持有)或打开失败 -> 返回 None。
进程异常退出时 OS 会自动释放该锁。
?? -> ??????????(?????????,??????/????);
????(???? worker ??)????? -> ?? None?
??????? OS ????????
"""
try:
lock_path.parent.mkdir(parents=True, exist_ok=True)
handle = open(lock_path, "a+")
except OSError:
logger.exception("Curator scheduler: 打开锁文件失败")
logger.exception("Curator scheduler: ???????")
return None
try:
if sys.platform == "win32":
@@ -367,7 +367,7 @@ def _acquire_singleton_lock(lock_path):
def _release_singleton_lock(handle) -> None:
"""释放并关闭 _acquire_singleton_lock 拿到的锁文件句柄。"""
"""????? _acquire_singleton_lock ?????????"""
try:
if sys.platform == "win32":
import msvcrt
@@ -388,24 +388,24 @@ def _release_singleton_lock(handle) -> None:
async def _curator_scheduler_loop() -> None:
"""常驻后台任务:每隔 _CURATOR_CHECK_INTERVAL_SECONDS 检查一次需不需要进化。
"""??????:?? _CURATOR_CHECK_INTERVAL_SECONDS ???????????
多 worker 部署下用文件锁做 leader 选举 —— 只有抢到锁的那个 worker 进程真正
运行调度,其余 worker 直接退出,避免 curator 被重复触发、重复调用 LLM。
被 cancel 时 CancelledError 会穿过 ``except Exception``(它是 BaseException
子类,不会被捕获),协程随之干净退出并在 finally 里释放锁。
? worker ???????? leader ?? ?? ???????? worker ????
????,?? worker ????,?? curator ?????????? LLM?
? cancel ? CancelledError ??? ``except Exception``(?? BaseException
??,?????),?????????? finally ?????
"""
from deerflow.config.runtime_paths import runtime_home
from deerflow.skills.curator import maybe_run_curator
lock_handle = _acquire_singleton_lock(runtime_home() / ".curator_scheduler.lock")
if lock_handle is None:
logger.info("Curator scheduler: 另一个 worker 已持有调度锁,本 worker 跳过")
logger.info("Curator scheduler: ??? worker ??????,? worker ??")
return
logger.info("Curator scheduler: 取得调度锁,本 worker 负责 curator 调度")
logger.info("Curator scheduler: ?????,? worker ?? curator ??")
try:
# 启动后延迟一会儿再首次检查,避开启动高峰
# ?????????????,??????
await asyncio.sleep(_CURATOR_STARTUP_DELAY_SECONDS)
while True:
try:
@@ -477,7 +477,7 @@ _LEADERBOARD_SNAPSHOT_STARTUP_DELAY_SECONDS = _env_float(
)
# Hour (local Beijing time, 0-23) after which the scheduler force-recomputes the
# *previous* day's snapshot once, so records that landed near midnight are fully
# captured. Default 1 → "每天凌晨 1 点重新统计前一天".
# captured. Default 1 ? "???? 1 ????????".
_LEADERBOARD_FINALIZE_HOUR = min(
23,
_env_int("ADMIN_LEADERBOARD_FINALIZE_HOUR", 1, minimum=0),
@@ -648,7 +648,7 @@ async def _claim_leaderboard_snapshot(
if age < running_ttl_seconds:
return False
row.status = "running"
row.message = "排行榜后台统计任务已认领。"
row.message = "?????????????"
row.error = None
row.updated_at = datetime.now(UTC)
await _time_leaderboard_scheduler_db_operation(
@@ -701,7 +701,7 @@ async def _maybe_finalize_previous_leaderboard_day(session_factory, settings_has
[yesterday],
settings_hash,
force=True,
message="每日凌晨重新统计前一天完整数据,确保全部记录纳入统计。",
message="???????????????????????????",
)
if queued:
logger.info(
@@ -734,10 +734,10 @@ async def _run_leaderboard_snapshot_tick() -> tuple[int, int]:
session_factory,
stat_dates,
settings_hash,
message="排行榜后台统计任务定期预热/补算队列。",
message="?????????????/?????",
)
# 每天凌晨(默认 1 点)后强制重算"前一天"的完整快照一次,覆盖跨午夜才落库的记录。
# ??????? 1 ???????"???"????????????????????
try:
await _maybe_finalize_previous_leaderboard_day(session_factory, settings_hash)
except Exception:
@@ -816,31 +816,31 @@ async def _leaderboard_snapshot_scheduler_loop() -> None:
def _warn_if_multi_worker() -> None:
"""检测 uvicorn / gunicorn 多 worker 部署并打 ERROR 日志。
"""?? uvicorn / gunicorn ? worker ???? ERROR ???
主 agent / canvas / AI 写作流式响应在 Gateway 进程内通过 ``StreamBridge``
维护 run_id → SSE 订阅者的内存 map。多 worker 下,用户在 worker A 发起
``/runs/stream``、稍后在 worker B 调 join/resume,会因 in-process map 互相
看不见而 404 / 卡死。
? agent / canvas / AI ??????? Gateway ????? ``StreamBridge``
?? run_id ? SSE ?????? map?? worker ????? worker A ??
``/runs/stream``???? worker B ? join/resume??? in-process map ??
???? 404 / ???
所以这里在 startup 打 ERROR 强提醒。**不抛异常 fail-fast** —— 万一
运维确实知道自己在干嘛(比如配了 sticky session),不应该阻止启动。
????? startup ? ERROR ????**???? fail-fast** ?? ??
???????????????? sticky session??????????
"""
# uvicorn / gunicorn 都不会主动暴露 worker 数到代码层。可以从环境变量
# 或者通过观察 sibling 进程数粗略判断。优先用 WEB_CONCURRENCY(gunicorn
# / uvicorn 都认)。
# uvicorn / gunicorn ??????? worker ?????????????
# ?????? sibling ??????????? WEB_CONCURRENCY?gunicorn
# / uvicorn ????
web_concurrency = os.environ.get("WEB_CONCURRENCY", "").strip()
if web_concurrency and web_concurrency.isdigit() and int(web_concurrency) > 1:
logger.error(
"⚠ 检测到 WEB_CONCURRENCY=%s(多 worker 部署)。AI 写作 / 主 agent "
"流式响应依赖进程内内存状态,多 worker 下会出现「会话丢失 / "
"SSE 断流 / resume 400」等问题。强烈建议改为 WEB_CONCURRENCY=1 "
"或在反向代理上启用 sticky session。详见 AIWritingPage.tsx 的注释。",
"? ??? WEB_CONCURRENCY=%s?? worker ????AI ?? / ? agent "
"??????????????? worker ????????? / "
"SSE ?? / resume 400??????????? WEB_CONCURRENCY=1 "
"????????? sticky session??? AIWritingPage.tsx ????",
web_concurrency,
)
return
# 兜底:扫 /proc 里同名 uvicorn 进程数(只在 Linux 容器里能用)
# ???? /proc ??? uvicorn ?????? Linux ??????
if not sys.platform.startswith("linux"):
return
try:
@@ -858,13 +858,13 @@ def _warn_if_multi_worker() -> None:
peers += 1
if peers > 1:
logger.error(
"⚠ 检测到 %d 个同名 uvicorn 进程。多 worker 部署会导致 AI 写作 "
"「会话丢失 / SSE 断流」等问题,请改为单 worker 或在反向代理上 "
"启用 sticky session。",
"? ??? %d ??? uvicorn ???? worker ????? AI ?? "
"????? / SSE ??????????? worker ??????? "
"?? sticky session?",
peers,
)
except Exception:
# 检测本身失败不算事 —— 这只是个提醒
# ????????? ?? ??????
pass
@@ -872,8 +872,8 @@ def _warn_if_multi_worker() -> None:
async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
"""Application lifespan handler."""
# 早一点检测多 worker 部署 —— 这是 AI 写作 / 主 agent 流式响应的常见
# 配置陷阱,每个 worker 启动时都会打一条 ERROR 强提醒运维
# ?????? worker ?? ?? ?? AI ?? / ? agent ???????
# ??????? worker ???????? ERROR ?????
_warn_if_multi_worker()
# Load config and check necessary environment variables at startup
@@ -895,7 +895,7 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
# Run Alembic migrations before anything else touches the schema.
await _run_db_migrations()
# 预热 recall_budget cache(从 DB 批量加载,使 get_effective_memory_config 同步可读)
# ?? recall_budget cache?? DB ????,? get_effective_memory_config ?????
try:
from deerflow.config.memory_config import load_recall_budget_cache
@@ -911,18 +911,18 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
# Must run AFTER langgraph_runtime so app.state.store is available for thread migration
await _ensure_admin_user(app)
# Migrate "ownerless" on-disk skills into the new SkillStore table —
# Migrate "ownerless" on-disk skills into the new SkillStore table ?
# makes the multitenancy switch transparent for existing deployments.
try:
await _sync_legacy_skills(app)
except Exception:
logger.exception("Skill ownership migration failed (non-fatal)")
# 圆桌功能型 agent(intent / recommender / coordinator)缺失自动补建。
# 必须在 _sync_legacy_agents **之前**——这样刚补建的目录能在同一次启动被
# 扫描并 upsert 进 agents 表,新机器一启动 /api/agents 就有这三个,
# 无需用户先打开圆桌页触发懒加载。.deer-flow/ 被 gitignore,新机器 clone
# 后这些目录不存在,seeder 从源码资源 _roundtable_seed_assets/ 复制验证版。
# ????? agent(intent / recommender / coordinator)???????
# ??? _sync_legacy_agents **??**??????????????????
# ??? upsert ? agents ?,?????? /api/agents ?????,
# ????????????????.deer-flow/ ? gitignore,??? clone
# ????????,seeder ????? _roundtable_seed_assets/ ??????
try:
from app.gateway.routers._roundtable_seed import (
ensure_roundtable_functional_agents,
@@ -934,9 +934,9 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
"Roundtable functional agent seeding failed at startup (non-fatal)"
)
# AI 写作功能型 agent(researcher / outliner / writer / editor)缺失自动补建。
# 同圆桌:必须在 _sync_legacy_agents 之前,补建目录才能在同一次启动被
# upsert 进 agents 表,admin 启动后即可在智能体管理页编辑四个写作 agent。
# AI ????? agent(researcher / outliner / writer / editor)???????
# ???:??? _sync_legacy_agents ??,?????????????
# upsert ? agents ?,admin ?????????????????? agent?
try:
from app.gateway.routers._ai_writing_seed import (
ensure_ai_writing_functional_agents,
@@ -948,7 +948,7 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
"AI-writing functional agent seeding failed at startup (non-fatal)"
)
# 业务链条管理:幂等补建固定业务行(3Q/6BF/7BF/8BF),agent/chain 初始为空。
# ?????????????????3Q/6BF/7BF/8BF??agent/chain ?????
try:
business_mapping_store = getattr(app.state, "business_mapping_store", None)
if business_mapping_store is not None:
@@ -958,10 +958,10 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
except Exception:
logger.exception("Business mapping seeding failed at startup (non-fatal)")
# 一次性把 .deer-flow/agents/{id}/ 目录里历史遗留的 agent 导入 agents 表。
# 路由层原本每次请求都跑一遍这个 sync(扫文件系统 + N 个 agent 各一次 DB
# SELECT/UPDATE),把它挪到启动期 + 路由层只保留 TTL 兜底,显著降低
# /api/agents 列表延迟。
# ???? .deer-flow/agents/{id}/ ???????? agent ?? agents ??
# ??????????????? sync(????? + N ? agent ??? DB
# SELECT/UPDATE),??????? + ?????? TTL ??,????
# /api/agents ?????
try:
from app.gateway.routers.agents import (
_mark_legacy_sync_fresh,
@@ -974,7 +974,7 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
await _sync_legacy_agents(agent_store)
_mark_legacy_sync_fresh()
# ``services.py`` keeps an independent TTL gate (used by
# ``start_run`` → ``_enforce_agent_access``). Without this
# ``start_run`` ? ``_enforce_agent_access``). Without this
# ack, the first ``/runs/stream`` after startup would
# re-run the heavy sync even though we just did it.
_services_mark_sync_fresh()
@@ -982,17 +982,17 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
except Exception:
logger.exception("Legacy agent sync failed at startup (non-fatal)")
# 启动 skill curator 调度器:常驻后台任务,每小时检查一次需不需要进化
# (per-user interval_hours 仍是真正决定要不要跑进化的闸)。
# ?? skill curator ???:??????,?????????????
# (per-user interval_hours ??????????????)?
try:
app.state.curator_scheduler_task = asyncio.create_task(_curator_scheduler_loop())
except Exception:
logger.exception("Skill curator scheduler failed to start (non-fatal)")
# 启动 AI 写作会话定期清理:按 config.yaml 的 ai_writing.cleanup 段调度,
# 默认每天凌晨 3 点扫一次,删 retention_days(默认 7 天)以前的会话,联动
# 清理 MySQL 业务表 + SQLite checkpointer 里的 thread 状态。多 worker
# 下用文件锁选 leader,只有抢到锁的 worker 真正跑清理。
# ?? AI ????????:? config.yaml ? ai_writing.cleanup ???,
# ?????? 3 ????,? retention_days(?? 7 ?)?????,??
# ?? MySQL ??? + SQLite checkpointer ?? thread ???? worker
# ?????? leader,?????? worker ??????
try:
from app.gateway.ai_writing_cleanup import cleanup_scheduler_loop
@@ -1031,7 +1031,7 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
yield
# 停止 curator 调度器
# ?? curator ???
curator_task = getattr(app.state, "curator_scheduler_task", None)
if curator_task is not None:
curator_task.cancel()
@@ -1042,7 +1042,7 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
except Exception:
logger.exception("Curator scheduler shutdown error")
# 停止 AI 写作清理调度器
# ?? AI ???????
cleanup_task = getattr(app.state, "ai_writing_cleanup_task", None)
if cleanup_task is not None:
cleanup_task.cancel()
@@ -1063,9 +1063,9 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
except Exception:
logger.exception("Leaderboard snapshot scheduler shutdown error")
# 阶段 7 删冗余:AI 写作 graph 不再自管 checkpointer / in-flight task;
# thread 状态由 LangGraph Server 主 checkpointer 接管,lifespan 出口由
# ``langgraph_runtime()`` 的 contextmanager 统一关闭,无需在这里特殊处理。
# ?? 7 ????AI ?? graph ???? checkpointer / in-flight task?
# thread ??? LangGraph Server ? checkpointer ???lifespan ???
# ``langgraph_runtime()`` ? contextmanager ???????????????
try:
scheduler_service = getattr(app.state, "scheduled_task_service", None)
@@ -1197,7 +1197,7 @@ This gateway provides custom endpoints for models, MCP configuration, skills, an
# CORS
if is_auth_disabled():
logger.warning(
"DEER_FLOW_AUTH_DISABLED is set — API authentication is bypassed. "
"DEER_FLOW_AUTH_DISABLED is set ? API authentication is bypassed. "
"Do not use this in production."
)
@@ -1222,7 +1222,7 @@ This gateway provides custom endpoints for models, MCP configuration, skills, an
if _cors_allow_any_origin():
logger.warning(
"Permissive CORS enabled — any Origin is echoed back (GATEWAY_CORS_ALLOW_ALL or "
"Permissive CORS enabled ? any Origin is echoed back (GATEWAY_CORS_ALLOW_ALL or "
"DEER_FLOW_AUTH_DISABLED). Do not expose publicly."
)
app.add_middleware(
@@ -1238,7 +1238,7 @@ This gateway provides custom endpoints for models, MCP configuration, skills, an
elif cors_origins:
if "*" in cors_origins:
logger.error(
"GATEWAY_CORS_ORIGINS contains '*' with allow_credentials=True — invalid. "
"GATEWAY_CORS_ORIGINS contains '*' with allow_credentials=True ? invalid. "
"Set GATEWAY_CORS_ALLOW_ALL=1 instead, or list explicit origins."
)
cors_origins = [o for o in cors_origins if o != "*"]
@@ -1319,10 +1319,10 @@ This gateway provides custom endpoints for models, MCP configuration, skills, an
# Roundtable-planning business chains at /api/roundtable-chains
app.include_router(roundtable_chains.router)
# Global business → agent / chain mapping at /api/business-mapping
# Global business ? agent / chain mapping at /api/business-mapping
app.include_router(business_mapping.router)
# Light-app registry (轻应用中心) at /api/light-apps
# Light-app registry (?????) at /api/light-apps
app.include_router(light_apps.router)
# Standalone parallel multi-agent panel: fan-out streaming + saved tasks
@@ -1400,7 +1400,7 @@ This gateway provides custom endpoints for models, MCP configuration, skills, an
# Public, unauthenticated read-only share snapshots (/api/public/shares/*).
app.include_router(thread_shares.public_router)
# 圆桌会商草稿分享(import 复制副本 / view 只读结果报告页)
# ?????????import ???? / view ????????
app.include_router(roundtable_draft_shares.router)
# Public, unauthenticated read-only roundtable result (/api/public/roundtable-shares/*).
app.include_router(roundtable_draft_shares.public_router)
@@ -1408,17 +1408,17 @@ This gateway provides custom endpoints for models, MCP configuration, skills, an
# Stateless Runs API (stream/wait without a pre-existing thread)
app.include_router(runs.router)
# --- 圆桌规划三步流(前端 /page/workspace/roundtable/planning)---
# 实现见 app/gateway/routers/{multi_agent,intent,recommend}.py;
# 文档见 frontend-web/docs/multi-agent-backend-dev.md
app.include_router(multi_agent.router) # Step 2:init(SSE) + run/stream(leader|special)
app.include_router(intent.router) # Step 1:/api/intent/init + /stream
app.include_router(recommend.router) # Step 1→2:/api/recommend/stream(无 init)
# Step 3 产物列表/下载复用下方 artifacts.router(/api/threads/{id}/artifacts)
# --- ?????????? /page/workspace/roundtable/planning?---
# ??? app/gateway/routers/{multi_agent,intent,recommend}.py?
# ??? frontend-web/docs/multi-agent-backend-dev.md
app.include_router(multi_agent.router) # Step 2?init(SSE) + run/stream(leader|special)
app.include_router(intent.router) # Step 1?/api/intent/init + /stream
app.include_router(recommend.router) # Step 1?2?/api/recommend/stream?? init?
# Step 3 ????/?????? artifacts.router?/api/threads/{id}/artifacts?
# 探活端点:无需鉴权(见 auth_middleware._PUBLIC_PATH_PREFIXES)。
# 同时暴露裸 ``/health`` 与带 API 前缀的 ``/api/health``,方便监控按
# 任意 base URL 直接探活,不会被认证中间件拦成 401。
# ?????????(? auth_middleware._PUBLIC_PATH_PREFIXES)?
# ????? ``/health`` ?? API ??? ``/api/health``,?????
# ?? base URL ????,?????????? 401?
@app.get("/health", tags=["health"])
@app.get("/api/health", tags=["health"])
async def health_check() -> dict: