939 lines
37 KiB
Python
939 lines
37 KiB
Python
import asyncio
|
||
import logging
|
||
import os
|
||
import queue
|
||
import sys
|
||
import threading
|
||
from collections.abc import AsyncGenerator
|
||
from contextlib import asynccontextmanager
|
||
|
||
from dotenv import load_dotenv
|
||
from fastapi import FastAPI
|
||
from fastapi.middleware.cors import CORSMiddleware
|
||
|
||
# Load .env BEFORE any module that reads os.environ at import time.
|
||
# Without this, GATEWAY_CORS_ORIGINS / AUTH_JWT_SECRET / DEER_FLOW_AUTH_DISABLED
|
||
# placed in /workspace/.env are silently ignored when the container is started
|
||
# without explicit `-e` flags, leading to ephemeral JWT secrets, missing CORS
|
||
# origins, and "Failed to fetch" in the browser.
|
||
load_dotenv()
|
||
|
||
from app.gateway.auth.disabled_mode import is_auth_disabled # noqa: E402
|
||
from app.gateway.auth_middleware import AuthMiddleware
|
||
from app.gateway.config import get_gateway_config
|
||
from app.gateway.csrf_middleware import CSRFMiddleware
|
||
from app.gateway.deps import langgraph_runtime
|
||
from app.gateway.routers import (
|
||
ai_writing,
|
||
admin_users,
|
||
agents,
|
||
artifacts,
|
||
assistants_compat,
|
||
auth,
|
||
channels,
|
||
feedback,
|
||
intent,
|
||
llm_metrics,
|
||
mcp,
|
||
memory,
|
||
models,
|
||
multi_agent,
|
||
notifications,
|
||
public_scheduled_tasks,
|
||
recommend,
|
||
recommended_questions,
|
||
roundtable_chains,
|
||
roundtable_drafts,
|
||
runs,
|
||
scheduled_tasks,
|
||
skills,
|
||
suggestions,
|
||
system_settings,
|
||
tags,
|
||
thread_runs,
|
||
tool_metrics,
|
||
thread_shares,
|
||
threads,
|
||
uploads,
|
||
user_prompts,
|
||
writing,
|
||
)
|
||
from deerflow.config import app_config as deerflow_app_config
|
||
from deerflow.config.app_config import apply_logging_level
|
||
from deerflow.runtime.scheduler import ScheduledTaskService, set_scheduled_task_service
|
||
|
||
AppConfig = deerflow_app_config.AppConfig
|
||
get_app_config = deerflow_app_config.get_app_config
|
||
|
||
# Default logging; lifespan overrides from config.yaml log_level.
|
||
logging.basicConfig(
|
||
level=logging.INFO,
|
||
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
|
||
datefmt="%Y-%m-%d %H:%M:%S",
|
||
)
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# Upper bound (seconds) each lifespan shutdown hook is allowed to run.
|
||
# Bounds worker exit time so uvicorn's reload supervisor does not keep
|
||
# firing signals into a worker that is stuck waiting for shutdown cleanup.
|
||
_SHUTDOWN_HOOK_TIMEOUT_SECONDS = 5.0
|
||
|
||
|
||
async def _ensure_admin_user(app: FastAPI) -> None:
|
||
"""Startup hook: handle first boot and migrate orphan threads otherwise.
|
||
|
||
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
|
||
authentication have existing LangGraph thread data that needs an
|
||
owner assigned.
|
||
First boot (no admin exists):
|
||
- Does NOT create any user accounts automatically.
|
||
- 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
|
||
existing LangGraph thread metadata that has no owner_id.
|
||
|
||
No SQL persistence migration is needed: the four user_id columns
|
||
(threads_meta, runs, run_events, feedback) only come into existence
|
||
alongside the auth module via create_all, so freshly created tables
|
||
never contain NULL-owner rows.
|
||
"""
|
||
from sqlalchemy import select
|
||
|
||
from app.gateway.deps import get_local_provider
|
||
from deerflow.persistence.engine import get_session_factory
|
||
from deerflow.persistence.user.model import UserRow
|
||
|
||
try:
|
||
provider = get_local_provider()
|
||
except RuntimeError:
|
||
# Auth persistence may not be initialized in some test/boot paths.
|
||
# Skip admin migration work rather than failing gateway startup.
|
||
logger.warning("Auth persistence not ready; skipping admin bootstrap check")
|
||
return
|
||
|
||
sf = get_session_factory()
|
||
if sf is None:
|
||
return
|
||
|
||
admin_count = await provider.count_admin_users()
|
||
|
||
if admin_count == 0:
|
||
logger.info("=" * 60)
|
||
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
|
||
# LangGraph thread metadata that pre-dates the auth module.
|
||
async with sf() as session:
|
||
stmt = select(UserRow).where(UserRow.system_role == "admin").limit(1)
|
||
row = (await session.execute(stmt)).scalar_one_or_none()
|
||
|
||
if row is None:
|
||
return # Should not happen (admin_count > 0 above), but be safe.
|
||
|
||
admin_id = str(row.id)
|
||
|
||
# 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:
|
||
try:
|
||
migrated = await _migrate_orphaned_threads(store, admin_id)
|
||
if migrated:
|
||
logger.info("Migrated %d orphan LangGraph thread(s) to admin", migrated)
|
||
except Exception:
|
||
logger.exception("LangGraph thread migration failed (non-fatal)")
|
||
|
||
|
||
async def _iter_store_items(store, namespace, *, page_size: int = 500):
|
||
"""Paginated async iterator over a LangGraph store namespace.
|
||
|
||
Replaces the old hardcoded ``limit=1000`` call with a cursor-style
|
||
loop so that environments with more than one page of orphans do
|
||
not silently lose data. Terminates when a page is empty OR when a
|
||
short page arrives (indicating the last page).
|
||
"""
|
||
offset = 0
|
||
while True:
|
||
batch = await store.asearch(namespace, limit=page_size, offset=offset)
|
||
if not batch:
|
||
return
|
||
for item in batch:
|
||
yield item
|
||
if len(batch) < page_size:
|
||
return
|
||
offset += page_size
|
||
|
||
|
||
async def _sync_legacy_skills(app: FastAPI) -> None:
|
||
"""Idempotently register existing on-disk ``skills/custom/*`` directories
|
||
in the ``skills`` DB table, owned by the first admin and published=True.
|
||
|
||
Without this step, after upgrading to per-user skill ownership the existing
|
||
custom skill folders would not be visible to anyone (the SkillStore
|
||
visibility filter relies on a DB row). Running this once on every boot is
|
||
cheap (folder scan + one ``ensure_legacy`` per name) and idempotent.
|
||
|
||
Non-fatal: a failure here MUST NOT block server startup.
|
||
"""
|
||
from sqlalchemy import select
|
||
|
||
from deerflow.persistence.engine import get_session_factory
|
||
from deerflow.persistence.user.model import UserRow
|
||
from deerflow.skills.storage import get_or_new_skill_storage
|
||
from deerflow.skills.types import SkillCategory
|
||
|
||
sf = get_session_factory()
|
||
if sf is None:
|
||
return
|
||
|
||
skill_store = getattr(app.state, "skill_store", None)
|
||
if skill_store is None:
|
||
return
|
||
|
||
async with sf() as session:
|
||
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
|
||
# claiming via the UI once they exist.
|
||
admin_id: str | None = None
|
||
else:
|
||
admin_id = str(row.id)
|
||
|
||
config = getattr(app.state, "config", None)
|
||
storage = get_or_new_skill_storage(app_config=config)
|
||
try:
|
||
skills = storage.load_skills(enabled_only=False)
|
||
except Exception:
|
||
logger.exception("_sync_legacy_skills: could not list on-disk skills")
|
||
return
|
||
|
||
inserted = 0
|
||
for skill in skills:
|
||
if skill.category != SkillCategory.CUSTOM:
|
||
continue
|
||
try:
|
||
existing = await skill_store.get_any(skill.name)
|
||
if existing is not None:
|
||
continue
|
||
await skill_store.ensure_legacy(
|
||
{"name": skill.name, "owner_user_id": admin_id, "published": True}
|
||
)
|
||
inserted += 1
|
||
except Exception:
|
||
logger.warning("_sync_legacy_skills: could not register %s", skill.name, exc_info=True)
|
||
if inserted:
|
||
logger.info("Skill ownership migration: registered %d legacy custom skill(s)", inserted)
|
||
|
||
|
||
async def _migrate_orphaned_threads(store, admin_user_id: str) -> int:
|
||
"""Migrate LangGraph store threads with no user_id to the given admin.
|
||
|
||
Uses cursor pagination so all orphans are migrated regardless of
|
||
count. Returns the number of rows migrated.
|
||
"""
|
||
migrated = 0
|
||
async for item in _iter_store_items(store, ("threads",)):
|
||
metadata = item.value.get("metadata", {})
|
||
if not metadata.get("user_id"):
|
||
metadata["user_id"] = admin_user_id
|
||
item.value["metadata"] = metadata
|
||
await store.aput(("threads",), item.key, item.value)
|
||
migrated += 1
|
||
return migrated
|
||
|
||
|
||
async def _run_db_migrations() -> None:
|
||
"""Run Alembic migrations at startup so schema changes are applied automatically."""
|
||
from pathlib import Path
|
||
|
||
from alembic import command
|
||
from alembic.config import Config
|
||
|
||
import deerflow.persistence as _persistence_pkg
|
||
from deerflow.persistence.engine import get_engine
|
||
|
||
engine = get_engine()
|
||
if engine is None:
|
||
logger.info("Persistence backend=memory — skipping Alembic migrations")
|
||
return
|
||
|
||
if os.getenv("DEER_FLOW_SKIP_ALEMBIC", "").lower() in {"1", "true", "yes"}:
|
||
logger.warning("DEER_FLOW_SKIP_ALEMBIC is set; skipping Alembic migrations")
|
||
return
|
||
|
||
db_url = engine.url.render_as_string(hide_password=False).replace("%", "%%")
|
||
migrations_dir = str(Path(_persistence_pkg.__file__).parent / "migrations")
|
||
result_queue: queue.Queue[BaseException | None] = queue.Queue(maxsize=1)
|
||
|
||
def _run_sync() -> None:
|
||
try:
|
||
cfg = Config()
|
||
cfg.set_main_option("script_location", migrations_dir)
|
||
cfg.set_main_option("sqlalchemy.url", db_url)
|
||
command.upgrade(cfg, "head")
|
||
except BaseException as exc:
|
||
result_queue.put(exc)
|
||
else:
|
||
result_queue.put(None)
|
||
|
||
try:
|
||
try:
|
||
timeout_s = float(os.getenv("DEER_FLOW_ALEMBIC_TIMEOUT", "60"))
|
||
except ValueError:
|
||
timeout_s = 60.0
|
||
|
||
worker = threading.Thread(target=_run_sync, name="deerflow-alembic-upgrade", daemon=True)
|
||
worker.start()
|
||
try:
|
||
deadline = asyncio.get_running_loop().time() + timeout_s
|
||
while result_queue.empty():
|
||
if asyncio.get_running_loop().time() >= deadline:
|
||
raise asyncio.TimeoutError
|
||
await asyncio.sleep(0.2)
|
||
result = result_queue.get_nowait()
|
||
except asyncio.TimeoutError:
|
||
logger.error(
|
||
"Database migration did not finish within %.0fs; continuing startup on create_all schema. "
|
||
"Likely cause: MySQL metadata-lock wait or a slow DDL. Set DEER_FLOW_SKIP_ALEMBIC=1 "
|
||
"to skip startup migrations, or inspect MySQL with SHOW FULL PROCESSLIST.",
|
||
timeout_s,
|
||
)
|
||
return
|
||
if result is not None:
|
||
raise result
|
||
logger.info("Database migrations applied (head)")
|
||
except Exception:
|
||
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 分钟
|
||
|
||
|
||
def _acquire_singleton_lock(lock_path):
|
||
"""跨平台非阻塞独占文件锁,用于多 worker 下的 leader 选举。
|
||
|
||
成功 -> 返回持有锁的文件句柄(调用方须保持其存活,直到进程退出/主动释放);
|
||
抢不到锁(已有其它 worker 持有)或打开失败 -> 返回 None。
|
||
进程异常退出时 OS 会自动释放该锁。
|
||
"""
|
||
try:
|
||
lock_path.parent.mkdir(parents=True, exist_ok=True)
|
||
handle = open(lock_path, "a+")
|
||
except OSError:
|
||
logger.exception("Curator scheduler: 打开锁文件失败")
|
||
return None
|
||
try:
|
||
if sys.platform == "win32":
|
||
import msvcrt
|
||
|
||
handle.seek(0)
|
||
msvcrt.locking(handle.fileno(), msvcrt.LK_NBLCK, 1)
|
||
else:
|
||
import fcntl
|
||
|
||
fcntl.flock(handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
|
||
except OSError:
|
||
handle.close()
|
||
return None
|
||
return handle
|
||
|
||
|
||
def _release_singleton_lock(handle) -> None:
|
||
"""释放并关闭 _acquire_singleton_lock 拿到的锁文件句柄。"""
|
||
try:
|
||
if sys.platform == "win32":
|
||
import msvcrt
|
||
|
||
handle.seek(0)
|
||
msvcrt.locking(handle.fileno(), msvcrt.LK_UNLCK, 1)
|
||
else:
|
||
import fcntl
|
||
|
||
fcntl.flock(handle.fileno(), fcntl.LOCK_UN)
|
||
except OSError:
|
||
pass
|
||
finally:
|
||
try:
|
||
handle.close()
|
||
except OSError:
|
||
pass
|
||
|
||
|
||
async def _curator_scheduler_loop() -> None:
|
||
"""常驻后台任务:每隔 _CURATOR_CHECK_INTERVAL_SECONDS 检查一次需不需要进化。
|
||
|
||
多 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 跳过")
|
||
return
|
||
|
||
logger.info("Curator scheduler: 取得调度锁,本 worker 负责 curator 调度")
|
||
try:
|
||
# 启动后延迟一会儿再首次检查,避开启动高峰
|
||
await asyncio.sleep(_CURATOR_STARTUP_DELAY_SECONDS)
|
||
while True:
|
||
try:
|
||
await maybe_run_curator()
|
||
except Exception:
|
||
logger.exception("Curator scheduler tick failed (non-fatal)")
|
||
await asyncio.sleep(_CURATOR_CHECK_INTERVAL_SECONDS)
|
||
finally:
|
||
_release_singleton_lock(lock_handle)
|
||
|
||
|
||
def _warn_if_multi_worker() -> None:
|
||
"""检测 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 / 卡死。
|
||
|
||
所以这里在 startup 打 ERROR 强提醒。**不抛异常 fail-fast** —— 万一
|
||
运维确实知道自己在干嘛（比如配了 sticky session），不应该阻止启动。
|
||
"""
|
||
# 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,
|
||
)
|
||
return
|
||
|
||
# 兜底：扫 /proc 里同名 uvicorn 进程数（只在 Linux 容器里能用）
|
||
if not sys.platform.startswith("linux"):
|
||
return
|
||
try:
|
||
from pathlib import Path
|
||
|
||
peers = 0
|
||
for entry in Path("/proc").iterdir():
|
||
if not entry.name.isdigit():
|
||
continue
|
||
try:
|
||
cmdline = (entry / "cmdline").read_bytes().replace(b"\x00", b" ").decode("utf-8", "ignore")
|
||
except OSError:
|
||
continue
|
||
if "uvicorn" in cmdline and "app.gateway.app" in cmdline:
|
||
peers += 1
|
||
if peers > 1:
|
||
logger.error(
|
||
"⚠ 检测到 %d 个同名 uvicorn 进程。多 worker 部署会导致 AI 写作 "
|
||
"「会话丢失 / SSE 断流」等问题，请改为单 worker 或在反向代理上 "
|
||
"启用 sticky session。",
|
||
peers,
|
||
)
|
||
except Exception:
|
||
# 检测本身失败不算事 —— 这只是个提醒
|
||
pass
|
||
|
||
|
||
@asynccontextmanager
|
||
async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
|
||
"""Application lifespan handler."""
|
||
|
||
# 早一点检测多 worker 部署 —— 这是 AI 写作 / 主 agent 流式响应的常见
|
||
# 配置陷阱，每个 worker 启动时都会打一条 ERROR 强提醒运维
|
||
_warn_if_multi_worker()
|
||
|
||
# Load config and check necessary environment variables at startup
|
||
try:
|
||
app.state.config = get_app_config()
|
||
apply_logging_level(app.state.config.log_level)
|
||
logger.info("Configuration loaded successfully")
|
||
except Exception as e:
|
||
error_msg = f"Failed to load configuration during gateway startup: {e}"
|
||
logger.exception(error_msg)
|
||
raise RuntimeError(error_msg) from e
|
||
config = get_gateway_config()
|
||
logger.info(f"Starting API Gateway on {config.host}:{config.port}")
|
||
|
||
# Initialize LangGraph runtime components (StreamBridge, RunManager, checkpointer, store)
|
||
async with langgraph_runtime(app):
|
||
logger.info("LangGraph runtime initialised")
|
||
|
||
# Run Alembic migrations before anything else touches the schema.
|
||
await _run_db_migrations()
|
||
|
||
# 预热 recall_budget cache（从 DB 批量加载,使 get_effective_memory_config 同步可读）
|
||
try:
|
||
from deerflow.config.memory_config import load_recall_budget_cache
|
||
|
||
_mc_store = getattr(app.state, "memory_config_store", None)
|
||
if _mc_store is not None:
|
||
_recall_budgets = await _mc_store.load_all()
|
||
load_recall_budget_cache(_recall_budgets)
|
||
logger.info("Memory config cache warmed: %d user override(s)", len(_recall_budgets))
|
||
except Exception:
|
||
logger.exception("Memory config cache warm-up failed (non-fatal)")
|
||
|
||
# Ensure admin user exists (auto-create on first boot)
|
||
# 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 —
|
||
# 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/ 复制验证版。
|
||
try:
|
||
from app.gateway.routers._roundtable_seed import (
|
||
ensure_roundtable_functional_agents,
|
||
)
|
||
|
||
ensure_roundtable_functional_agents()
|
||
except Exception:
|
||
logger.exception(
|
||
"Roundtable functional agent seeding failed at startup (non-fatal)"
|
||
)
|
||
|
||
# 一次性把 .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,
|
||
_sync_legacy_agents,
|
||
)
|
||
from app.gateway.services import _mark_legacy_sync_fresh as _services_mark_sync_fresh
|
||
|
||
agent_store = getattr(app.state, "agent_store", None)
|
||
if agent_store is not 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
|
||
# ack, the first ``/runs/stream`` after startup would
|
||
# re-run the heavy sync even though we just did it.
|
||
_services_mark_sync_fresh()
|
||
logger.info("Legacy folder-backed agents synced at startup")
|
||
except Exception:
|
||
logger.exception("Legacy agent sync failed at startup (non-fatal)")
|
||
|
||
# 启动 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 真正跑清理。
|
||
try:
|
||
from app.gateway.ai_writing_cleanup import cleanup_scheduler_loop
|
||
|
||
app.state.ai_writing_cleanup_task = asyncio.create_task(cleanup_scheduler_loop(app))
|
||
except Exception:
|
||
logger.exception("AI writing cleanup scheduler failed to start (non-fatal)")
|
||
|
||
# Start IM channel service if any channels are configured
|
||
try:
|
||
from app.channels.service import start_channel_service
|
||
|
||
channel_service = await start_channel_service(app.state.config)
|
||
logger.info("Channel service started: %s", channel_service.get_status())
|
||
except Exception:
|
||
logger.exception("No IM channels configured or channel service failed to start")
|
||
|
||
scheduled_tasks_enabled = os.environ.get("SCHEDULED_TASKS_ENABLED", "true").strip().lower()
|
||
if scheduled_tasks_enabled in {"0", "false", "no", "off"}:
|
||
logger.info("Scheduled task service disabled by SCHEDULED_TASKS_ENABLED=%s", scheduled_tasks_enabled)
|
||
else:
|
||
try:
|
||
scheduled_task_store = getattr(app.state, "scheduled_task_store", None)
|
||
if scheduled_task_store is not None:
|
||
scheduler_service = ScheduledTaskService(app, scheduled_task_store)
|
||
app.state.scheduled_task_service = scheduler_service
|
||
set_scheduled_task_service(scheduler_service)
|
||
await scheduler_service.start()
|
||
logger.info("Scheduled task service started")
|
||
except Exception:
|
||
logger.exception("Scheduled task service failed to start")
|
||
|
||
yield
|
||
|
||
# 停止 curator 调度器
|
||
curator_task = getattr(app.state, "curator_scheduler_task", None)
|
||
if curator_task is not None:
|
||
curator_task.cancel()
|
||
try:
|
||
await curator_task
|
||
except asyncio.CancelledError:
|
||
pass
|
||
except Exception:
|
||
logger.exception("Curator scheduler shutdown error")
|
||
|
||
# 停止 AI 写作清理调度器
|
||
cleanup_task = getattr(app.state, "ai_writing_cleanup_task", None)
|
||
if cleanup_task is not None:
|
||
cleanup_task.cancel()
|
||
try:
|
||
await cleanup_task
|
||
except asyncio.CancelledError:
|
||
pass
|
||
except Exception:
|
||
logger.exception("AI writing cleanup scheduler shutdown error")
|
||
|
||
# 阶段 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)
|
||
if scheduler_service is not None:
|
||
await scheduler_service.stop()
|
||
except Exception:
|
||
logger.exception("Failed to stop scheduled task service")
|
||
finally:
|
||
set_scheduled_task_service(None)
|
||
|
||
# Stop channel service on shutdown (bounded to prevent worker hang)
|
||
try:
|
||
from app.channels.service import stop_channel_service
|
||
|
||
await asyncio.wait_for(
|
||
stop_channel_service(),
|
||
timeout=_SHUTDOWN_HOOK_TIMEOUT_SECONDS,
|
||
)
|
||
except TimeoutError:
|
||
logger.warning(
|
||
"Channel service shutdown exceeded %.1fs; proceeding with worker exit.",
|
||
_SHUTDOWN_HOOK_TIMEOUT_SECONDS,
|
||
)
|
||
except Exception:
|
||
logger.exception("Failed to stop channel service")
|
||
|
||
logger.info("Shutting down API Gateway")
|
||
|
||
|
||
def create_app() -> FastAPI:
|
||
"""Create and configure the FastAPI application.
|
||
|
||
Returns:
|
||
Configured FastAPI application instance.
|
||
"""
|
||
config = get_gateway_config()
|
||
docs_kwargs = {"docs_url": "/docs", "redoc_url": "/redoc", "openapi_url": "/openapi.json"} if config.enable_docs else {"docs_url": None, "redoc_url": None, "openapi_url": None}
|
||
|
||
app = FastAPI(
|
||
title="DeerFlow API Gateway",
|
||
description="""
|
||
## DeerFlow API Gateway
|
||
|
||
API Gateway for DeerFlow - A LangGraph-based AI agent backend with sandbox execution capabilities.
|
||
|
||
### Features
|
||
|
||
- **Models Management**: Query and retrieve available AI models
|
||
- **MCP Configuration**: Manage Model Context Protocol (MCP) server configurations
|
||
- **Memory Management**: Access and manage global memory data for personalized conversations
|
||
- **Skills Management**: Query and manage skills and their enabled status
|
||
- **Artifacts**: Access thread artifacts and generated files
|
||
- **Health Monitoring**: System health check endpoints
|
||
|
||
### Architecture
|
||
|
||
LangGraph requests are handled by nginx reverse proxy.
|
||
This gateway provides custom endpoints for models, MCP configuration, skills, and artifacts.
|
||
""",
|
||
version="0.1.0",
|
||
lifespan=lifespan,
|
||
**docs_kwargs,
|
||
openapi_tags=[
|
||
{
|
||
"name": "models",
|
||
"description": "Operations for querying available AI models and their configurations",
|
||
},
|
||
{
|
||
"name": "mcp",
|
||
"description": "Manage Model Context Protocol (MCP) server configurations",
|
||
},
|
||
{
|
||
"name": "memory",
|
||
"description": "Access and manage global memory data for personalized conversations",
|
||
},
|
||
{
|
||
"name": "skills",
|
||
"description": "Manage skills and their configurations",
|
||
},
|
||
{
|
||
"name": "artifacts",
|
||
"description": "Access and download thread artifacts and generated files",
|
||
},
|
||
{
|
||
"name": "uploads",
|
||
"description": "Upload and manage user files for threads",
|
||
},
|
||
{
|
||
"name": "threads",
|
||
"description": "Manage DeerFlow thread-local filesystem data",
|
||
},
|
||
{
|
||
"name": "agents",
|
||
"description": "Create and manage custom agents with per-agent config and prompts",
|
||
},
|
||
{
|
||
"name": "suggestions",
|
||
"description": "Generate follow-up question suggestions for conversations",
|
||
},
|
||
{
|
||
"name": "channels",
|
||
"description": "Manage IM channel integrations (Feishu, Slack, Telegram)",
|
||
},
|
||
{
|
||
"name": "assistants-compat",
|
||
"description": "LangGraph Platform-compatible assistants API (stub)",
|
||
},
|
||
{
|
||
"name": "runs",
|
||
"description": "LangGraph Platform-compatible runs lifecycle (create, stream, cancel)",
|
||
},
|
||
{
|
||
"name": "scheduled-tasks",
|
||
"description": "Create, manage, and execute user scheduled agent tasks",
|
||
},
|
||
{
|
||
"name": "health",
|
||
"description": "Health check and system status endpoints",
|
||
},
|
||
],
|
||
)
|
||
|
||
# Auth: reject unauthenticated requests to non-public paths (fail-closed safety net)
|
||
app.add_middleware(AuthMiddleware)
|
||
|
||
# CSRF: Double Submit Cookie pattern for state-changing requests
|
||
app.add_middleware(CSRFMiddleware)
|
||
|
||
# CORS
|
||
if is_auth_disabled():
|
||
logger.warning(
|
||
"DEER_FLOW_AUTH_DISABLED is set — API authentication is bypassed. "
|
||
"Do not use this in production."
|
||
)
|
||
|
||
def _cors_allow_any_origin() -> bool:
|
||
"""Mirror any browser Origin (works with credentials); cannot use literal '*'."""
|
||
v = os.environ.get("GATEWAY_CORS_ALLOW_ALL", "").strip().lower()
|
||
return v in ("1", "true", "yes", "on") or is_auth_disabled()
|
||
|
||
cors_origins_env = os.environ.get("GATEWAY_CORS_ORIGINS", "").strip()
|
||
cors_origins: list[str] = [o.strip() for o in cors_origins_env.split(",") if o.strip()] if cors_origins_env else []
|
||
|
||
# Headers the LangGraph SDK / browser clients need to read from cross-origin
|
||
# responses. Without exposing them, the SDK's `response.headers.get(...)` returns
|
||
# null on cross-origin streaming responses, breaking run-id extraction and reconnect.
|
||
cors_expose_headers = [
|
||
"Content-Disposition",
|
||
"Content-Location",
|
||
"Location",
|
||
"X-Run-Id",
|
||
"X-Thread-Id",
|
||
]
|
||
|
||
if _cors_allow_any_origin():
|
||
logger.warning(
|
||
"Permissive CORS enabled — any Origin is echoed back (GATEWAY_CORS_ALLOW_ALL or "
|
||
"DEER_FLOW_AUTH_DISABLED). Do not expose publicly."
|
||
)
|
||
app.add_middleware(
|
||
CORSMiddleware,
|
||
allow_origins=[],
|
||
# fullmatch() on the Origin header; browsers send scheme://host[:port]
|
||
allow_origin_regex=r".+",
|
||
allow_credentials=True,
|
||
allow_methods=["*"],
|
||
allow_headers=["*"],
|
||
expose_headers=cors_expose_headers,
|
||
)
|
||
elif cors_origins:
|
||
if "*" in cors_origins:
|
||
logger.error(
|
||
"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 != "*"]
|
||
if cors_origins:
|
||
app.add_middleware(
|
||
CORSMiddleware,
|
||
allow_origins=cors_origins,
|
||
allow_credentials=True,
|
||
allow_methods=["*"],
|
||
allow_headers=["*"],
|
||
expose_headers=cors_expose_headers,
|
||
)
|
||
else:
|
||
# Safe-by-default fallback for local frontend development. Without any CORS
|
||
# middleware, browser preflight OPTIONS requests will hit routes directly and
|
||
# fail with 405, which blocks login and all JSON POST APIs.
|
||
dev_origins = [
|
||
"http://localhost:5173",
|
||
"http://127.0.0.1:5173",
|
||
]
|
||
logger.warning(
|
||
"No CORS configuration found; enabling local dev origins only: %s. "
|
||
"Set GATEWAY_CORS_ORIGINS or GATEWAY_CORS_ALLOW_ALL for deployment.",
|
||
", ".join(dev_origins),
|
||
)
|
||
app.add_middleware(
|
||
CORSMiddleware,
|
||
allow_origins=dev_origins,
|
||
allow_credentials=True,
|
||
allow_methods=["*"],
|
||
allow_headers=["*"],
|
||
expose_headers=cors_expose_headers,
|
||
)
|
||
|
||
# Include routers
|
||
# Models API is mounted at /api/models
|
||
app.include_router(models.router)
|
||
|
||
# MCP API is mounted at /api/mcp
|
||
app.include_router(mcp.router)
|
||
|
||
# Memory API is mounted at /api/memory
|
||
app.include_router(memory.router)
|
||
|
||
# Skills API is mounted at /api/skills
|
||
app.include_router(skills.router)
|
||
|
||
# Artifacts API is mounted at /api/threads/{thread_id}/artifacts
|
||
app.include_router(artifacts.router)
|
||
|
||
# Uploads API is mounted at /api/threads/{thread_id}/uploads
|
||
app.include_router(uploads.router)
|
||
|
||
# Thread cleanup API is mounted at /api/threads/{thread_id}
|
||
app.include_router(threads.router)
|
||
|
||
# Agents API is mounted at /api/agents
|
||
app.include_router(agents.router)
|
||
|
||
# Recommended Questions API is mounted at /api/recommended-questions
|
||
app.include_router(recommended_questions.router)
|
||
|
||
app.include_router(user_prompts.router)
|
||
|
||
# Roundtable-planning drafts + recommendation history at /api/roundtable-drafts
|
||
app.include_router(roundtable_drafts.router)
|
||
|
||
# Roundtable-planning business chains at /api/roundtable-chains
|
||
app.include_router(roundtable_chains.router)
|
||
|
||
# Suggestions API is mounted at /api/threads/{thread_id}/suggestions
|
||
app.include_router(suggestions.router)
|
||
|
||
# Writing rewrite API is mounted at /api/writing
|
||
app.include_router(writing.router)
|
||
|
||
# AI Writing multi-agent API is mounted at /api/ai-writing
|
||
app.include_router(ai_writing.router)
|
||
|
||
# Scheduled task API is mounted at /api/scheduled-tasks
|
||
app.include_router(scheduled_tasks.router)
|
||
|
||
# Public, unauthenticated read-only scheduled task API
|
||
# (/api/public/scheduled-tasks/*). Lets external systems query tasks and
|
||
# surface results without a DeerFlow login.
|
||
app.include_router(public_scheduled_tasks.router)
|
||
|
||
# System settings API is mounted at /api/system-settings
|
||
app.include_router(system_settings.router)
|
||
|
||
# Admin LLM metrics audit + export
|
||
app.include_router(llm_metrics.router)
|
||
|
||
# Admin window into other users' threads / messages / memory
|
||
app.include_router(admin_users.router)
|
||
|
||
# Admin tool / skill call metrics audit + export
|
||
app.include_router(tool_metrics.router)
|
||
|
||
# Channels API is mounted at /api/channels
|
||
app.include_router(channels.router)
|
||
|
||
# Assistants compatibility API (LangGraph Platform stub)
|
||
app.include_router(assistants_compat.router)
|
||
|
||
# Auth API is mounted at /api/v1/auth
|
||
app.include_router(auth.router)
|
||
|
||
# Feedback API is mounted at /api/threads/{thread_id}/runs/{run_id}/feedback
|
||
app.include_router(feedback.router)
|
||
|
||
# Notifications API is mounted at /api/notifications
|
||
app.include_router(notifications.router)
|
||
|
||
# Tags API is mounted at /api/tags
|
||
app.include_router(tags.router)
|
||
|
||
# Thread Runs API (LangGraph Platform-compatible runs lifecycle)
|
||
app.include_router(thread_runs.router)
|
||
|
||
app.include_router(thread_shares.router)
|
||
# Public, unauthenticated read-only share snapshots (/api/public/shares/*).
|
||
app.include_router(thread_shares.public_router)
|
||
|
||
# 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）
|
||
|
||
@app.get("/health", tags=["health"])
|
||
async def health_check() -> dict:
|
||
"""Health check endpoint.
|
||
|
||
Returns:
|
||
Service health status information.
|
||
"""
|
||
return {"status": "healthy", "service": "deer-flow-gateway"}
|
||
|
||
return app
|
||
|
||
|
||
# Create app instance for uvicorn
|
||
app = create_app()
|