From f358cfb7c7147b4c9f7b9a3cdb89843bdede71c6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E8=91=A3=E5=AE=8F=E5=87=AF?= Date: Wed, 20 May 2026 12:51:04 +0800 Subject: [PATCH] =?UTF-8?q?=E5=AE=B9=E9=94=99=E5=A4=84=E7=90=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../src/open-canvas/pages/AIWritingPage.tsx | 83 +++++ .../backend/app/gateway/ai_writing_cleanup.py | 300 ++++++++++++++++++ .../backend/app/gateway/app.py | 113 +++++++ .../backend/app/gateway/routers/ai_writing.py | 244 +++++++++++++- .../deerflow/agents/ai_writing/__init__.py | 16 +- .../deerflow/agents/ai_writing/graph.py | 200 +++++++++++- .../persistence/ai_writing_sessions/sql.py | 37 ++- offline-backend-20260512/config.yaml | 16 +- 8 files changed, 988 insertions(+), 21 deletions(-) create mode 100644 offline-backend-20260512/backend/app/gateway/ai_writing_cleanup.py diff --git a/frontend-web/src/open-canvas/pages/AIWritingPage.tsx b/frontend-web/src/open-canvas/pages/AIWritingPage.tsx index d2ebe25c..ba2ad6e5 100644 --- a/frontend-web/src/open-canvas/pages/AIWritingPage.tsx +++ b/frontend-web/src/open-canvas/pages/AIWritingPage.tsx @@ -1,3 +1,86 @@ +/** + * AI 写作入口页（左右双栏：聊天面板 + 草稿预览）。 + * + * ────────────────────────────────────────────────────────────────────────── + * Docker 内网部署后「会话丢失 / transcript 报 500 / resume 报 400」排查指南 + * ────────────────────────────────────────────────────────────────────────── + * + * 后端涉及接口（前缀 /api/ai-writing）： + * POST /start 启动写作流程 + * GET /{session_id}/stream SSE 进度流 + * POST /{session_id}/resume 提交用户干预 → 继续流程 + * PUT /sessions/{session_id}/transcript 保存对话时间线 + * + * ── 1) 确认 alembic 迁移已跑过（修 transcript 500）───────────────────── + * transcript 列在 MySQL 上原是 TEXT(64 KB 上限)，长时间线会溢出抛 500。 + * 迁移 20260519_02 把它改成 LONGTEXT。容器启动会自动跑，验证方法： + * + * docker exec sh -c "cd /app/backend \ + * && PYTHONPATH=. uv run alembic current" + * docker exec mysql -uroot -p deerflow -e \ + * "SHOW COLUMNS FROM ai_writing_sessions LIKE 'transcript';" + * + * 列类型应是 longtext。若仍是 text，手动执行： + * docker exec sh -c "cd /app/backend \ + * && PYTHONPATH=. uv run alembic upgrade head" + * + * ── 2) 确认 .deer-flow 数据卷已挂出（修「容器没重启也丢会话」）────── + * AI 写作 graph 的 checkpointer 已从 MemorySaver 换成 AsyncSqliteSaver， + * 文件落到 .deer-flow/data/checkpoints.db。docker-compose 必须把该目录 + * 挂出来，否则容器重启数据照样丢： + * + * services: + * gateway: + * volumes: + * - ./data:/app/.deer-flow # 关键：挂出 .deer-flow + * + * ── 3) 确认 worker 数 = 1（关键嫌疑：多 worker 内存隔离）──────────── + * _event_queues / _session_pending_pause 都是模块级 dict，每个 worker + * 一份。多 worker 下，/start 命中 worker A、/resume 命中 worker B，B + * 的 dict 没有这个 session → 404/「会话丢失」。 + * + * docker exec ps -ef | grep -E 'uvicorn|gunicorn' + * + * 若 > 1：要么把启动参数改成 --workers 1，要么在 nginx 加 sticky + * session（ip_hash / cookie），要么把队列搬 Redis。 + * + * ── 4) 排查节点异常（嫌疑：graph 节点抛错触发清理分支）───────────── + * _stream_graph_segment 的 except 分支会清掉 _event_queues[session_id]， + * 用户下一次 resume 就 404。日志里搜： + * + * docker logs 2>&1 | grep -E \ + * 'ai_writing graph segment failed|ai_writing.*error' | tail -50 + * + * ── 5) 排查反向代理掐 SSE 长连接（嫌疑：proxy_read_timeout 太短）─── + * SSE 默认连接挂半个小时以上，nginx 默认 60s 会断。检查内网反代： + * + * grep -rE 'proxy_read_timeout|sse|keepalive' /etc/nginx/ + * + * 需在 SSE 路由上加：proxy_read_timeout 3600s; proxy_buffering off; + * + * ── 定期清理（每天 03:00 跑）──────────────────────────────────────── + * config.yaml -> ai_writing.cleanup 段（retention_days: 7）。联动清理 + * MySQL 业务表 + SQLite checkpointer 里的 thread 状态。手动验证： + * + * docker exec mysql -uroot -p deerflow -e \ + * "SELECT COUNT(*) FROM ai_writing_sessions \ + * WHERE updated_at < NOW() - INTERVAL 7 DAY;" + * + * ── 不想等凌晨 3 点？手动触发一次清理（admin only）────────────────── + * POST /api/ai-writing/admin/cleanup + * Body（全部可选，留空走 config 默认）： + * { "retention_days": 7, "only_finished": false, "batch_size": 500 } + * 返回： + * { "deleted_db": N, "deleted_checkpoints": N, + * "elapsed_seconds": X, "params": {...} } + * + * 例：清掉所有 1 小时前的会话（测试用）： + * curl -X POST http:///api/ai-writing/admin/cleanup \ + * -H 'Content-Type: application/json' \ + * -H 'Cookie: ' \ + * -d '{"retention_days": 0}' + * ────────────────────────────────────────────────────────────────────────── + */ import { useRef } from 'react'; import { TooltipProvider } from '@/components/ui/tooltip'; import type { GroupImperativeHandle } from 'react-resizable-panels'; diff --git a/offline-backend-20260512/backend/app/gateway/ai_writing_cleanup.py b/offline-backend-20260512/backend/app/gateway/ai_writing_cleanup.py new file mode 100644 index 00000000..78673a7b --- /dev/null +++ b/offline-backend-20260512/backend/app/gateway/ai_writing_cleanup.py @@ -0,0 +1,300 @@ +"""AI 写作会话定期清理。 + +按 ``config.yaml`` 的 ``ai_writing.cleanup`` 段做联动清理： + + 1. 找出 MySQL ``ai_writing_sessions`` 表里 ``updated_at < now - retention_days`` + 的会话 id 列表（分批，避免大事务） + 2. 对每个 id 调用 LangGraph checkpointer 的 ``adelete_thread()``，删掉 + SQLite checkpointer 里对应的 thread 状态 + 3. 再把 MySQL 行批量删掉 + +整个流程「先删 checkpointer 再删 DB」是有意为之 —— 即便清理被中途中断， +也只会出现「checkpointer 删了但 DB 没删」的孤儿，下一轮清理会发现这些 +DB 行仍然过期，重新尝试清理（checkpointer 删一个不存在的 thread 是 idempotent）。 +反过来如果先删 DB，万一 checkpointer 删失败，文件里就永久留着不可达的状态。 + +多 worker 下用文件锁做 leader election，只有抢到锁的 worker 才会真正跑清理。 +""" +from __future__ import annotations + +import asyncio +import logging +import sys +from datetime import UTC, datetime, timedelta +from pathlib import Path + +from croniter import croniter + +logger = logging.getLogger(__name__) + + +# ── 文件锁 ──────────────────────────────────────────────────────────────────── + + +def _acquire_lock(lock_path: Path): + """跨平台非阻塞独占文件锁，多 worker 下做 leader election。 + + 成功 → 返回持有锁的文件句柄（调用方需保持其存活直到主动释放）； + 抢不到 / 打开失败 → 返回 None。进程退出时 OS 自动释放。 + """ + try: + lock_path.parent.mkdir(parents=True, exist_ok=True) + handle = open(lock_path, "a+") + except OSError: + logger.exception("ai_writing_cleanup: 打开锁文件失败 path=%s", lock_path) + 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_lock(handle) -> None: + 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 _delete_checkpoint_thread(checkpointer, thread_id: str) -> None: + """删 checkpointer 里 thread_id 对应的全部状态。失败只记日志，不抛。""" + if checkpointer is None: + return + # LangGraph 异步删法首选 adelete_thread；老版本只有同步 delete_thread。 + delete_async = getattr(checkpointer, "adelete_thread", None) + if delete_async is not None: + try: + await delete_async(thread_id) + return + except Exception: + logger.warning( + "ai_writing_cleanup: adelete_thread failed thread_id=%s", thread_id, exc_info=True + ) + return + delete_sync = getattr(checkpointer, "delete_thread", None) + if delete_sync is not None: + try: + # 同步方法跑在线程池里，避免堵 event loop + await asyncio.to_thread(delete_sync, thread_id) + except Exception: + logger.warning( + "ai_writing_cleanup: delete_thread failed thread_id=%s", thread_id, exc_info=True + ) + + +def _get_active_session_ids() -> set[str]: + """返回内存里**当前正在跑**的 session_id 集合。 + + 这些会话的写作流程还活着 —— 任何清理都不能动它们，否则用户「正在写」 + 的稿子会被秒删，体验灾难。延迟到下一轮清理（24 小时后）再处理也比误 + 删强。 + """ + try: + # 延迟 import 避免循环依赖：cleanup 模块在 lifespan 启动时被 import， + # routers/ai_writing 也被 include_router 主动 import，两者拓扑无环但 + # 还是放函数里更稳。 + from app.gateway.routers.ai_writing import _event_queues + + return set(_event_queues.keys()) + except Exception: + # 兜底：拿不到就当作没有活动会话（最差就是清掉一个刚跑完的会话， + # 比误删正在写的要好得多 —— 用户至少能从 DB 历史里看到结果） + logger.warning("ai_writing_cleanup: 读取活动会话列表失败", exc_info=True) + return set() + + +async def run_cleanup_once( + *, + retention_days: int, + only_finished: bool = False, + batch_size: int = 500, +) -> dict: + """执行一轮清理。返回统计信息。 + + **安全保证**：清理永远跳过内存里仍在跑的会话（``_event_queues`` 里有 + 对应 key 的）—— 这条规则即使在 admin 手动触发清理时也强制生效，防止 + 把用户「正在写」的稿子误删。 + + 幂等 + 可重入 —— 多次调用对结果只会越扫越干净。即便清理途中被中断，下次 + 再跑会重新把 DB 里仍然过期的行扫到（checkpointer 删一个已不存在的 thread + 是 no-op）。 + """ + from deerflow.agents.ai_writing.graph import get_ai_writing_checkpointer + from deerflow.persistence.ai_writing_sessions import AIWritingSessionRepository + from deerflow.persistence.engine import get_session_factory + + sf = get_session_factory() + if sf is None: + logger.info("ai_writing_cleanup: persistence backend=memory，跳过清理") + return {"deleted_db": 0, "deleted_checkpoints": 0, "skipped_active": 0} + + repo = AIWritingSessionRepository(sf) + checkpointer = get_ai_writing_checkpointer() + active_ids = _get_active_session_ids() + + cutoff = datetime.now(UTC) - timedelta(days=retention_days) + logger.info( + "ai_writing_cleanup: 开始清理 cutoff=%s only_finished=%s batch_size=%d active=%d", + cutoff.isoformat(), + only_finished, + batch_size, + len(active_ids), + ) + + total_db = 0 + total_cp = 0 + total_skipped = 0 + while True: + ids = await repo.list_older_than(cutoff, only_finished=only_finished, limit=batch_size) + if not ids: + break + + # 关键保护：跳过任何仍在内存里跑的会话 + ids_to_delete = [sid for sid in ids if sid not in active_ids] + skipped = len(ids) - len(ids_to_delete) + if skipped: + logger.warning( + "ai_writing_cleanup: 跳过 %d 个仍在跑的会话: %s", + skipped, + [sid for sid in ids if sid in active_ids], + ) + total_skipped += skipped + + if not ids_to_delete: + # 本批全是活动会话 —— 但 list_older_than 是按 updated_at 升序的， + # 下一批拿出来的更旧，所以这种情况只可能整个 cutoff 区间都被活动 + # 会话占满，不用继续往下扫，break + break + + # 先删 checkpointer，再删 DB —— 保证「DB 删了 → checkpointer 一定也清干净」 + for sid in ids_to_delete: + await _delete_checkpoint_thread(checkpointer, sid) + total_cp += 1 + + deleted = await repo.delete_by_ids(ids_to_delete) + total_db += deleted + logger.info( + "ai_writing_cleanup: 本批删 %d 行（DB），%d 个 thread（checkpointer），跳过 %d", + deleted, + len(ids_to_delete), + skipped, + ) + + # 整批比 batch_size 小 → 没有更多了 + if len(ids) < batch_size: + break + + logger.info( + "ai_writing_cleanup: 清理完成 total_db=%d total_checkpoints=%d skipped_active=%d", + total_db, + total_cp, + total_skipped, + ) + return { + "deleted_db": total_db, + "deleted_checkpoints": total_cp, + "skipped_active": total_skipped, + } + + +# ── 调度循环 ────────────────────────────────────────────────────────────────── + + +def _next_cron_delay_seconds(cron_expr: str, base: datetime | None = None) -> float: + """根据 cron 表达式算到下一次触发的秒数。 + + 用 ``datetime.now(UTC)`` 做基准。croniter 接受 naive 时间，这里转成 naive + UTC 喂进去；当然如果用户希望按本地时区跑（凌晨 3 点），需要在外层调整 —— + 我们用本地时区基准，等价于 cron 在本地时区跑。 + """ + base = base or datetime.now() + nxt = croniter(cron_expr, base).get_next(datetime) + return max(1.0, (nxt - base).total_seconds()) + + +async def cleanup_scheduler_loop() -> None: + """常驻后台任务：按 cron 调度跑清理。 + + 多 worker 部署下文件锁做 leader election，没拿到锁的 worker 直接退出。 + 被 cancel 时 CancelledError 不会被 except Exception 吞掉，能干净退出。 + """ + from deerflow.config.app_config import get_app_config + from deerflow.config.runtime_paths import runtime_home + + try: + cfg = get_app_config() + except Exception: + logger.exception("ai_writing_cleanup: 加载 app config 失败，调度器退出") + return + + raw = getattr(cfg, "ai_writing", None) or {} + cleanup_cfg = (raw.get("cleanup") if isinstance(raw, dict) else None) or {} + if not cleanup_cfg.get("enabled", False): + logger.info("ai_writing_cleanup: cleanup.enabled=false，调度器不启动") + return + + cron_expr = str(cleanup_cfg.get("cron", "0 3 * * *")) + retention_days = int(cleanup_cfg.get("retention_days", 7)) + only_finished = bool(cleanup_cfg.get("delete_only_finished", False)) + batch_size = int(cleanup_cfg.get("batch_size", 500)) + + # 提前校验 cron 表达式，配错就早 fail 早暴露 + try: + croniter(cron_expr, datetime.now()) + except Exception: + logger.exception("ai_writing_cleanup: 非法 cron 表达式 %r，调度器退出", cron_expr) + return + + lock_handle = _acquire_lock(runtime_home() / ".ai_writing_cleanup.lock") + if lock_handle is None: + logger.info("ai_writing_cleanup: 另一个 worker 已持有调度锁，本 worker 跳过") + return + + logger.info( + "ai_writing_cleanup: 已启动 cron=%r retention_days=%d only_finished=%s batch_size=%d", + cron_expr, + retention_days, + only_finished, + batch_size, + ) + try: + while True: + delay = _next_cron_delay_seconds(cron_expr) + logger.info("ai_writing_cleanup: 下一次清理 %.0f 秒后", delay) + await asyncio.sleep(delay) + try: + await run_cleanup_once( + retention_days=retention_days, + only_finished=only_finished, + batch_size=batch_size, + ) + except Exception: + logger.exception("ai_writing_cleanup: 本轮清理失败（不影响下一轮）") + finally: + _release_lock(lock_handle) diff --git a/offline-backend-20260512/backend/app/gateway/app.py b/offline-backend-20260512/backend/app/gateway/app.py index b1556806..1c4a6851 100644 --- a/offline-backend-20260512/backend/app/gateway/app.py +++ b/offline-backend-20260512/backend/app/gateway/app.py @@ -70,6 +70,12 @@ logger = logging.getLogger(__name__) # firing signals into a worker that is stuck waiting for shutdown cleanup. _SHUTDOWN_HOOK_TIMEOUT_SECONDS = 5.0 +# AI 写作 in-flight task 收尾的最大等待时间。一个 graph 片段通常在几秒内 +# 跑到下一个 interrupt 或结束，但单段草稿生成走完整个 LLM 可能耗时较久， +# 给 30 秒兜底；超时则 cancel 强制结束，已落盘的 checkpoint 仍可用于重启 +# 后恢复，丢的只是最后一段 in-flight 但未 checkpoint 的进度。 +_AI_WRITING_DRAIN_TIMEOUT_SECONDS = 30.0 + async def _ensure_admin_user(app: FastAPI) -> None: """Startup hook: handle first boot and migrate orphan threads otherwise. @@ -361,10 +367,68 @@ async def _curator_scheduler_loop() -> None: _release_singleton_lock(lock_handle) +def _warn_if_multi_worker() -> None: + """检测 uvicorn / gunicorn 多 worker 部署并打 ERROR 日志。 + + AI 写作 + 主 agent 流式响应都依赖**进程内**的内存状态字典 + （``_event_queues``、``_session_pending_pause`` 等）。多 worker 下， + 用户的 /start /stream /resume 会被 OS 随机分配到不同 worker， + per-worker 的内存字典互相看不见 → SSE 事件丢失、会话「凭空消失」、 + resume 走不通。 + + 所以这里在 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() @@ -402,6 +466,17 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]: 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()) + 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 @@ -439,6 +514,44 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]: 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") + + # 等待所有 in-flight 写作 graph task 跑到下一个 interrupt / 结束。 + # 这是「写作流程能完整继续」的关键 —— 如果直接关 checkpointer + # connection，正在跑 astream 的 task 会在写 checkpoint 时抛错， + # 已经生成但还没落盘的进度全丢。 + try: + from app.gateway.routers.ai_writing import wait_for_inflight_graph_tasks + + await wait_for_inflight_graph_tasks(timeout=_AI_WRITING_DRAIN_TIMEOUT_SECONDS) + except Exception: + logger.exception("Wait for in-flight AI writing tasks failed (non-fatal)") + + # 关闭 AI 写作 graph 持有的 checkpointer 连接（aiosqlite Connection / pg pool） + try: + from deerflow.agents.ai_writing.graph import shutdown_ai_writing_graph + + await asyncio.wait_for( + shutdown_ai_writing_graph(), + timeout=_SHUTDOWN_HOOK_TIMEOUT_SECONDS, + ) + except TimeoutError: + logger.warning( + "AI writing checkpointer shutdown exceeded %.1fs; proceeding with worker exit.", + _SHUTDOWN_HOOK_TIMEOUT_SECONDS, + ) + except Exception: + logger.exception("Failed to shutdown AI writing checkpointer") + try: scheduler_service = getattr(app.state, "scheduled_task_service", None) if scheduler_service is not None: diff --git a/offline-backend-20260512/backend/app/gateway/routers/ai_writing.py b/offline-backend-20260512/backend/app/gateway/routers/ai_writing.py index 38acd5e3..968c6408 100644 --- a/offline-backend-20260512/backend/app/gateway/routers/ai_writing.py +++ b/offline-backend-20260512/backend/app/gateway/routers/ai_writing.py @@ -70,6 +70,28 @@ class RenameSessionRequest(BaseModel): title: str +class AdminCleanupRequest(BaseModel): + """手动清理接口的可选覆盖参数。 + + 全部留空 → 用 config.yaml 的 ai_writing.cleanup 段默认值；显式传值 + → 覆盖配置（一般用于运维想验证一个临时阈值，例如清掉「最近 1 小时 + 以前的」会话来模拟整轮调度）。 + """ + retention_days: Optional[int] = Field(default=None, ge=0, description="保留多少天以内的会话；0 = 不保留") + only_finished: Optional[bool] = Field(default=None, description="True 时只清 done/error 状态") + batch_size: Optional[int] = Field(default=None, ge=1, le=10000, description="单批最多删多少行") + + +class AdminCleanupResponse(BaseModel): + deleted_db: int + deleted_checkpoints: int + # 因「仍在跑」被保护掉的会话数。永远跳过 _event_queues 里活跃的会话， + # 即使 retention_days=0 也不会秒杀用户正在写的稿。 + skipped_active: int = 0 + elapsed_seconds: float + params: dict + + class SaveTranscriptRequest(BaseModel): # 前端组装的完整对话时间线（progressEvents + completedInterventions） transcript: dict @@ -153,11 +175,40 @@ _session_pending_pause: dict[str, tuple] = {} # lost session don't each rebuild a different queue object. _recovery_lock = asyncio.Lock() +# 在飞的 graph 推进任务（每次 /start 和 /resume 都会塞一个进来）。 +# Lifespan shutdown 时会等这些 task 跑到下一个 interrupt / 结束，再关闭 +# checkpointer 连接 —— 否则中途关 connection，task 内部的 sqlite 写入会抛 +# 错，导致 LLM 已经生成但还没落 checkpoint 的进度丢失。 +_inflight_graph_tasks: set[asyncio.Task] = set() + # ── 工具函数 ────────────────────────────────────────────────────────────────── def _sse(event: str, data: dict) -> str: - return f"event: {event}\ndata: {json.dumps(data, ensure_ascii=False)}\n\n" + """组装一帧 SSE。 + + JSON 序列化对 progress_events 来说理论上不会失败（节点里塞的都是 str/dict + /list/int 这类基础类型），但一旦塞进了 datetime / bytes / 自定义对象， + ``json.dumps`` 会抛 TypeError → SSE 生成器爆掉 → 客户端再也收不到任何 + 事件，包括关键的 await_user 和 done。 + + 这里用 ``default=str`` 兜底任何「带 __str__」的对象；如果连 ``str()`` + 都炸了，最后还兜一次手写一条 error 帧推出去 —— 写作流程「能继续」比 + 「漂亮地序列化」更重要。 + """ + try: + payload = json.dumps(data, ensure_ascii=False, default=str) + except Exception as exc: + logger.exception("ai_writing: SSE payload 序列化失败 event=%s", event) + payload = json.dumps( + { + "session_id": data.get("session_id") if isinstance(data, dict) else None, + "message": f"事件序列化失败（已降级）: {type(exc).__name__}", + }, + ensure_ascii=False, + ) + event = "notice" # 不要硬塞回原 event，万一是 error 会污染语义 + return f"event: {event}\ndata: {payload}\n\n" def _ai_writing_config() -> dict: @@ -193,14 +244,81 @@ def _truncate_title(text: str, max_len: int = 50) -> str: return text[:max_len] + ("…" if len(text) > max_len else "") +def _spawn_graph_task(coro) -> asyncio.Task: + """启一个 graph 推进 task 并登记到 _inflight_graph_tasks。 + + 用 ``add_done_callback`` 在完成时自动反登记 —— 不管 task 正常结束、抛 + 异常、还是被 cancel，都会清理；这样 _inflight_graph_tasks 不会泄漏。 + """ + task = asyncio.create_task(coro) + _inflight_graph_tasks.add(task) + task.add_done_callback(_inflight_graph_tasks.discard) + return task + + +async def wait_for_inflight_graph_tasks(timeout: float) -> tuple[int, int]: + """Lifespan shutdown 入口：阻塞等待所有 in-flight graph task 收尾。 + + 每个 ``_stream_graph_segment`` 跑到下一个 ``interrupt()`` 或 ``END`` 就 + 会自然返回；通常只需要等几秒到几十秒。超时后剩下的会被 ``cancel()``， + 但 cancel 完仍要等它们 ack 退出（带极短的 grace），否则下一步关 + checkpointer 时还会撞上活的协程。 + + Returns: + (drained, cancelled): 自然结束的数量、被 cancel 强制结束的数量 + """ + if not _inflight_graph_tasks: + return 0, 0 + + pending = set(_inflight_graph_tasks) + initial = len(pending) + logger.info("ai_writing: 等待 %d 个 in-flight graph task 收尾（timeout=%.1fs）", initial, timeout) + + done, still_running = await asyncio.wait(pending, timeout=timeout) + drained = len(done) + + cancelled = 0 + if still_running: + logger.warning( + "ai_writing: 仍有 %d 个 task 超时未结束，强制 cancel", len(still_running) + ) + for t in still_running: + t.cancel() + cancelled += 1 + # 给 cancel 一点 ack 时间，避免下游关连接时撞活协程 + await asyncio.wait(still_running, timeout=2.0) + + logger.info("ai_writing: in-flight 收尾完成 drained=%d cancelled=%d", drained, cancelled) + return drained, cancelled + + # ── 路由 ────────────────────────────────────────────────────────────────────── @router.post("/start", response_model=StartWritingResponse) async def start_writing(body: StartWritingRequest, request: Request): - """启动写作流程，返回 session_id。""" - from deerflow.agents.ai_writing.graph import AI_WRITING_GRAPH + """启动写作流程，返回 session_id。 + + **失败兜底契约**：本接口除非 Pydantic 校验 body 失败（422），其他任何 + 后端组件（DB / checkpointer / 文章类型仓库 / 用户解析）失败都不会导致 + /start 返回 5xx。哪怕兜底走的是「极简内存模式」也要先把 session_id + 交给前端，让它能起 SSE 接到 error 事件，避免 UI 卡死在 loading 状态。 + """ + from deerflow.agents.ai_writing.graph import get_ai_writing_graph from app.gateway.deps import get_ai_writing_session_store, get_article_type_store, get_current_user + # get_ai_writing_graph() 内部已对 sqlite/postgres 失败做了 InMemorySaver + # 降级，理论上永不抛。但 build_ai_writing_graph 拼图结构也可能因下游 + # 节点 import 失败抛错 —— 这种情况 graph 完全不可用，必须早 fail，前端 + # 弹错处理。 + try: + graph = await get_ai_writing_graph() + except Exception as e: + logger.exception("ai_writing: graph 构建彻底失败，无法启动") + raise HTTPException( + status_code=503, + detail=f"AI 写作服务暂不可用: {type(e).__name__}", + ) from e + session_id = str(uuid.uuid4()) queue: asyncio.Queue = asyncio.Queue() _event_queues[session_id] = queue @@ -258,7 +376,7 @@ async def start_writing(body: StartWritingRequest, request: Request): } config = _make_config(session_id, body.model_name, valid_article_types=valid_article_types) - asyncio.create_task(_stream_graph_segment(session_id, initial_state, config, AI_WRITING_GRAPH)) + _spawn_graph_task(_stream_graph_segment(session_id, initial_state, config, graph)) return StartWritingResponse(session_id=session_id) @@ -430,9 +548,11 @@ async def stream_progress(session_id: str): @router.post("/{session_id}/resume") async def resume_writing(session_id: str, body: ResumeRequest, request: Request): """用户提交干预决策，恢复暂停的流程。""" - from deerflow.agents.ai_writing.graph import AI_WRITING_GRAPH + from deerflow.agents.ai_writing.graph import get_ai_writing_graph from app.gateway.deps import get_current_user + graph = await get_ai_writing_graph() + if session_id not in _event_queues: # 内存会话注册表丢失（进程内清理后重连等）。先尝试从 checkpointer 恢复， # 让用户能在干预点继续，而不是直接 404 中断整个写作流程。 @@ -482,8 +602,8 @@ async def resume_writing(session_id: str, body: ResumeRequest, request: Request) # Clear pending pause so SSE reconnect won't re-inject the stale await_user event _session_pending_pause.pop(session_id, None) - asyncio.create_task( - _stream_graph_segment(session_id, Command(resume=intervention), config, AI_WRITING_GRAPH) + _spawn_graph_task( + _stream_graph_segment(session_id, Command(resume=intervention), config, graph) ) return {"message": "已提交，流程恢复中"} @@ -491,15 +611,112 @@ async def resume_writing(session_id: str, body: ResumeRequest, request: Request) @router.get("/{session_id}/state") async def get_state(session_id: str): """查询当前流程状态快照。""" - from deerflow.agents.ai_writing.graph import AI_WRITING_GRAPH + from deerflow.agents.ai_writing.graph import get_ai_writing_graph + graph = await get_ai_writing_graph() config = _make_config(session_id) - snapshot = await AI_WRITING_GRAPH.aget_state(config) + snapshot = await graph.aget_state(config) if not snapshot or not snapshot.values: raise HTTPException(status_code=404, detail="Session not found") return snapshot.values +# ── 管理接口 ────────────────────────────────────────────────────────────────── + +async def _require_admin(request: Request) -> None: + """放行 admin / auth-disabled，其余 403。 + + 与 scheduled_tasks._require_admin 同套语义：auth_disabled 模式下 + get_optional_user_from_request 返回 None，此时放行（开发环境一致体验）。 + """ + from app.gateway.deps import get_optional_user_from_request + + user = await get_optional_user_from_request(request) + if user is None: + return + if getattr(user, "system_role", None) != "admin": + raise HTTPException(status_code=403, detail="AI 写作清理管理接口仅限管理员") + + +def _resolve_cleanup_params(override: AdminCleanupRequest) -> dict: + """合并 config.yaml 默认值 + 请求显式覆盖。 + + 显式传 0 / False 也算覆盖（用 ``is not None`` 判定），方便测「保留 0 天 = + 全删」这种极端场景。 + """ + from deerflow.config import get_app_config + + raw = getattr(get_app_config(), "ai_writing", None) or {} + cfg = (raw.get("cleanup") if isinstance(raw, dict) else None) or {} + + return { + "retention_days": ( + override.retention_days if override.retention_days is not None + else int(cfg.get("retention_days", 7)) + ), + "only_finished": ( + override.only_finished if override.only_finished is not None + else bool(cfg.get("delete_only_finished", False)) + ), + "batch_size": ( + override.batch_size if override.batch_size is not None + else int(cfg.get("batch_size", 500)) + ), + } + + +@router.post("/admin/cleanup", response_model=AdminCleanupResponse) +async def admin_cleanup( + body: AdminCleanupRequest, + request: Request, +) -> AdminCleanupResponse: + """**仅管理员** 手动触发一次会话清理（不等凌晨 3 点定时调度）。 + + 用途： + - 验证 cleanup 配置改动是否生效 + - 内网紧急清理（磁盘满 / MySQL 行数过多） + - 调试：传 ``retention_days: 0`` 一刷拉清空，方便重新跑回归 + + 所有参数留空 → 沿用 config.yaml 的 ``ai_writing.cleanup`` 段默认。显式 + 传值会覆盖配置但**不修改 config.yaml**（一次性参数，下次定时调度还是用 + 配置文件的值）。 + """ + import time + + from app.gateway.ai_writing_cleanup import run_cleanup_once + + await _require_admin(request) + params = _resolve_cleanup_params(body) + + started = time.monotonic() + try: + result = await run_cleanup_once(**params) + except Exception as e: + logger.exception("ai_writing admin cleanup failed: params=%s", params) + # 抛 500 是合理的 —— 管理员触发清理失败需要看到失败原因，跟「业务 + # 接口被失败拖累」是两回事 + raise HTTPException( + status_code=500, + detail=f"清理执行失败: {type(e).__name__}: {e}", + ) from e + + elapsed = time.monotonic() - started + logger.info( + "ai_writing admin cleanup done: params=%s deleted_db=%d deleted_cp=%d elapsed=%.2fs", + params, + result["deleted_db"], + result["deleted_checkpoints"], + elapsed, + ) + return AdminCleanupResponse( + deleted_db=result["deleted_db"], + deleted_checkpoints=result["deleted_checkpoints"], + skipped_active=result.get("skipped_active", 0), + elapsed_seconds=round(elapsed, 3), + params=params, + ) + + # ── 内部辅助 ────────────────────────────────────────────────────────────────── async def _stream_graph_segment( @@ -650,13 +867,14 @@ def _event_type(evt: dict) -> str: async def _load_graph_snapshot(session_id: str): """读取 AI 写作 Graph 在该 session 上的状态快照（失败/无状态返回 None）。 - 注意：Graph 用的是内存 checkpointer（MemorySaver），所以这里只能在 - **进程存活**期间恢复；进程/容器重启后快照同样会丢失。 + Graph 的 checkpointer 现在按 config.yaml 持久化到磁盘（默认 SQLite）， + 所以容器重启后只要数据卷还在，仍然能恢复出来。 """ try: - from deerflow.agents.ai_writing.graph import AI_WRITING_GRAPH + from deerflow.agents.ai_writing.graph import get_ai_writing_graph - snapshot = await AI_WRITING_GRAPH.aget_state(_make_config(session_id)) + graph = await get_ai_writing_graph() + snapshot = await graph.aget_state(_make_config(session_id)) except Exception: logger.exception("ai_writing: aget_state failed during recovery: session=%s", session_id) return None diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/__init__.py b/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/__init__.py index f626726d..ae2fc5ac 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/__init__.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/__init__.py @@ -1,3 +1,15 @@ -from .graph import AI_WRITING_GRAPH, build_ai_writing_graph +from .graph import ( + AI_WRITING_GRAPH, + build_ai_writing_graph, + get_ai_writing_checkpointer, + get_ai_writing_graph, + shutdown_ai_writing_graph, +) -__all__ = ["AI_WRITING_GRAPH", "build_ai_writing_graph"] +__all__ = [ + "AI_WRITING_GRAPH", + "build_ai_writing_graph", + "get_ai_writing_graph", + "get_ai_writing_checkpointer", + "shutdown_ai_writing_graph", +] diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/graph.py b/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/graph.py index cb36382a..a4c593cc 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/graph.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/graph.py @@ -1,7 +1,8 @@ """AI 写作 LangGraph Graph，含 4 个用户干预点。""" from __future__ import annotations -from langgraph.checkpoint.memory import MemorySaver +import logging + from langgraph.graph import END, StateGraph from langgraph.types import interrupt @@ -12,6 +13,8 @@ from .nodes.writer_outline import writer_outline_node from .nodes.writer_draft import writer_draft_node from .nodes.editor import editor_node +logger = logging.getLogger(__name__) + # ── 干预节点 ────────────────────────────────────────────────────────────────── async def pause_material_confirm(state: AIWritingState) -> dict: @@ -298,7 +301,16 @@ def after_review_pause(state: AIWritingState) -> str: # ── 组装 Graph ──────────────────────────────────────────────────────────────── -def build_ai_writing_graph(): +def build_ai_writing_graph(checkpointer=None): + """构建并 compile AI 写作 graph。 + + Args: + checkpointer: 可选的 LangGraph checkpointer。 + - ``None`` → 退回 ``MemorySaver`` —— 仅供测试 / 开发使用。 + 进程或容器重启后会话状态全部丢失，前端拿着旧 session_id 调 + ``/resume`` 会得到 404。 + - 其他 → 传入的 (Async)BaseCheckpointSaver 实例，用于持久化。 + """ g = StateGraph(AIWritingState) g.add_node("intent_parser", intent_parser_node) @@ -339,8 +351,188 @@ def build_ai_writing_graph(): g.add_conditional_edges("pause_review", after_review_pause, {"end": END, "writer_draft": "writer_draft"}) - checkpointer = MemorySaver() + if checkpointer is None: + # MemorySaver 仅作为 fallback —— 提示一下，正式部署应通过 + # get_ai_writing_graph() 拿到带持久化 checkpointer 的 graph。 + from langgraph.checkpoint.memory import MemorySaver + + checkpointer = MemorySaver() return g.compile(checkpointer=checkpointer) -AI_WRITING_GRAPH = build_ai_writing_graph() +# ── 持久化 checkpointer 单例（懒加载）───────────────────────────────────────── + +# 模块级 graph 槽位。import 时为 None；首次 ``get_ai_writing_graph()`` 调用时 +# 按 config.yaml 的 checkpointer 段懒构建出真正的 graph。 +# 测试可通过 ``monkeypatch.setattr(graph_mod, "AI_WRITING_GRAPH", stub)`` +# 直接注入桩；本模块所有访问点都走 ``get_ai_writing_graph()`` 这一入口，会优先 +# 返回 module attribute 的值，保留测试桩注入路径。 +AI_WRITING_GRAPH = None + +# 持有 async checkpointer（底层 aiosqlite Connection 或 psycopg pool 的所有者）。 +# 进程生命周期内常驻；shutdown 时由 ``shutdown_ai_writing_graph`` 显式关闭。 +_async_checkpointer_singleton = None + + +def _fallback_to_memory_saver(reason: str): + """退回 InMemorySaver 的统一兜底入口。 + + 把 reason 打到 warning 日志里方便运维定位。InMemorySaver 进程重启会丢 + 状态，但**当次写作能跑完** —— 我们的首要目标是「写作流程能继续」， + 持久化是锦上添花。 + """ + from langgraph.checkpoint.memory import InMemorySaver + + logger.warning( + "ai_writing: 持久化 checkpointer 不可用，退回 InMemorySaver。原因: %s", + reason, + ) + return InMemorySaver() + + +async def _get_async_checkpointer(): + """按 config.yaml 的 checkpointer 段返回一个 async checkpointer 实例。 + + **绝不抛异常** —— 任何失败一律退回 InMemorySaver。这是为了保证「写作 + 流程能继续」这个首要目标：持久化坏了用户至少还能写完当次稿，重启会丢 + 会话但不影响当下创作。 + + 优先级： + 1. ``checkpointer:`` 段（``type=sqlite|postgres|memory``） + 2. fallback 到 ``InMemorySaver`` 并打 warning + + 返回的 saver 与主 agent 公用同一 sqlite 文件 —— LangGraph 的 schema + 用 ``(thread_id, checkpoint_ns)`` 做主键，AI 写作的 thread_id 是 + 会话 uuid，跟主 agent 的 LangGraph thread_id 不会撞。 + """ + global _async_checkpointer_singleton + if _async_checkpointer_singleton is not None: + return _async_checkpointer_singleton + + try: + from deerflow.config.checkpointer_config import get_checkpointer_config + + cp_cfg = get_checkpointer_config() + except Exception: + logger.exception("ai_writing: 加载 checkpointer 配置失败") + _async_checkpointer_singleton = _fallback_to_memory_saver("加载 config 抛异常") + return _async_checkpointer_singleton + + if cp_cfg is None: + _async_checkpointer_singleton = _fallback_to_memory_saver( + "config.yaml 未配置 checkpointer 段" + ) + return _async_checkpointer_singleton + + if cp_cfg.type == "memory": + from langgraph.checkpoint.memory import InMemorySaver + + _async_checkpointer_singleton = InMemorySaver() + logger.info("ai_writing: 配置 type=memory，使用 InMemorySaver") + return _async_checkpointer_singleton + + if cp_cfg.type == "sqlite": + try: + import aiosqlite + from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver + from deerflow.runtime.store._sqlite_utils import ( + ensure_sqlite_parent_dir, + resolve_sqlite_conn_str, + ) + + conn_str = resolve_sqlite_conn_str( + cp_cfg.connection_string or "ai_writing_checkpoints.db" + ) + ensure_sqlite_parent_dir(conn_str) + # 直接持有 aiosqlite.Connection —— 进程生命周期，shutdown 时显式关。 + conn = await aiosqlite.connect(conn_str) + saver = AsyncSqliteSaver(conn) + await saver.setup() + _async_checkpointer_singleton = saver + logger.info("ai_writing: 使用 AsyncSqliteSaver (%s)", conn_str) + return saver + except Exception as exc: + # sqlite 失败原因可能是：依赖未装 / 磁盘满 / 文件权限 / 路径不存在 / + # 文件被外部进程独占锁。任何一个都不能让 AI 写作整个不可用。 + logger.exception( + "ai_writing: AsyncSqliteSaver 初始化失败，conn_str=%r", + getattr(cp_cfg, "connection_string", None), + ) + _async_checkpointer_singleton = _fallback_to_memory_saver( + f"sqlite 初始化失败: {type(exc).__name__}: {exc}" + ) + return _async_checkpointer_singleton + + if cp_cfg.type == "postgres": + try: + from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver + from psycopg_pool import AsyncConnectionPool + + if not cp_cfg.connection_string: + raise ValueError("checkpointer.connection_string 是 postgres 必填项") + pool = AsyncConnectionPool(cp_cfg.connection_string, open=False) + await pool.open() + saver = AsyncPostgresSaver(pool) + await saver.setup() + _async_checkpointer_singleton = saver + logger.info("ai_writing: 使用 AsyncPostgresSaver") + return saver + except Exception as exc: + logger.exception("ai_writing: AsyncPostgresSaver 初始化失败") + _async_checkpointer_singleton = _fallback_to_memory_saver( + f"postgres 初始化失败: {type(exc).__name__}: {exc}" + ) + return _async_checkpointer_singleton + + # 未知 type —— config 配错也不能让服务挂 + _async_checkpointer_singleton = _fallback_to_memory_saver( + f"未知 checkpointer.type={cp_cfg.type!r}" + ) + return _async_checkpointer_singleton + + +async def get_ai_writing_graph(): + """懒加载并返回 AI 写作 graph 单例。 + + 第一次调用按 config 选定的 checkpointer 构建 graph 并缓存；后续调用直接 + 返回缓存。测试可在调用前通过 ``monkeypatch.setattr`` 直接替换模块级 + ``AI_WRITING_GRAPH`` —— 本函数会优先返回已设的值。 + """ + global AI_WRITING_GRAPH + if AI_WRITING_GRAPH is not None: + return AI_WRITING_GRAPH + checkpointer = await _get_async_checkpointer() + AI_WRITING_GRAPH = build_ai_writing_graph(checkpointer=checkpointer) + return AI_WRITING_GRAPH + + +def get_ai_writing_checkpointer(): + """返回当前已实例化的 checkpointer（清理任务用它来删 thread）。 + + 若尚未懒构建过则返回 None。 + """ + return _async_checkpointer_singleton + + +async def shutdown_ai_writing_graph() -> None: + """关闭 saver 持有的底层连接，并清空 graph 单例。Lifespan shutdown 调用。""" + global AI_WRITING_GRAPH, _async_checkpointer_singleton + saver = _async_checkpointer_singleton + _async_checkpointer_singleton = None + AI_WRITING_GRAPH = None + if saver is None: + return + # aiosqlite saver: saver.conn 是 aiosqlite.Connection + conn = getattr(saver, "conn", None) + if conn is not None and hasattr(conn, "close"): + try: + await conn.close() + except Exception: + logger.warning("ai_writing: 关闭 checkpointer connection 失败", exc_info=True) + # postgres saver: saver.pool 是 AsyncConnectionPool + pool = getattr(saver, "pool", None) + if pool is not None and hasattr(pool, "close"): + try: + await pool.close() + except Exception: + logger.warning("ai_writing: 关闭 checkpointer pool 失败", exc_info=True) diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/ai_writing_sessions/sql.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/ai_writing_sessions/sql.py index 48fd54ea..b6c4306e 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/ai_writing_sessions/sql.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/ai_writing_sessions/sql.py @@ -5,7 +5,7 @@ from __future__ import annotations import json from datetime import UTC, datetime -from sqlalchemy import select +from sqlalchemy import delete, select from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker from deerflow.persistence.ai_writing_sessions.model import AIWritingSessionRow @@ -160,3 +160,38 @@ class AIWritingSessionRepository: await session.delete(row) await session.commit() return True + + async def list_older_than( + self, + cutoff: datetime, + *, + only_finished: bool = False, + limit: int = 500, + ) -> list[str]: + """列出 updated_at 早于 cutoff 的会话 id（最多 limit 个）。 + + 清理任务先 list 出 id 列表，再用这些 id 联动删 checkpointer + 里对应的 thread，最后再删 MySQL 行 —— 保证两边一致。 + + Args: + cutoff: 截止时间，updated_at < cutoff 的会话被认为「过期」 + only_finished: True 时只清 status in (done/error)，False 不看状态 + limit: 单批最多返回多少条，用于分批处理避免大事务 + """ + stmt = select(AIWritingSessionRow.id).where(AIWritingSessionRow.updated_at < cutoff) + if only_finished: + stmt = stmt.where(AIWritingSessionRow.status.in_(("done", "error"))) + stmt = stmt.order_by(AIWritingSessionRow.updated_at.asc()).limit(limit) + async with self._sf() as session: + result = await session.execute(stmt) + return [row[0] for row in result.all()] + + async def delete_by_ids(self, session_ids: list[str]) -> int: + """按 id 列表批量删除。返回实际删除的行数。""" + if not session_ids: + return 0 + stmt = delete(AIWritingSessionRow).where(AIWritingSessionRow.id.in_(session_ids)) + async with self._sf() as session: + result = await session.execute(stmt) + await session.commit() + return int(result.rowcount or 0) diff --git a/offline-backend-20260512/config.yaml b/offline-backend-20260512/config.yaml index 9a1b006b..c132ddc0 100644 --- a/offline-backend-20260512/config.yaml +++ b/offline-backend-20260512/config.yaml @@ -117,8 +117,22 @@ database: pool_size: 10 echo_sql: false checkpointer: + # SQLite 文件路径用相对路径，由 resolve_sqlite_conn_str 解析成绝对路径。 + # Docker 部署：把容器内 .deer-flow 目录挂卷出来即可持久化（撑住容器重启）。 + # 这里同时被「主 agent」和「AI 写作 graph」共用 —— LangGraph 用 + # (thread_id, checkpoint_ns) 做隔离，session_id 是 uuid，不会撞。 type: sqlite - connection_string: D:\cursor-work\offline-backend-20260512\backend\.deer-flow\data\deerflow.db + connection_string: .deer-flow/data/checkpoints.db +# AI 写作子系统的会话清理配置。 +# 业务表(ai_writing_sessions @ MySQL) + checkpointer(checkpoints.db @ SQLite) 联动清理。 +# 留意：清理走 cron 表达式调度，默认每天凌晨 3 点跑一次，按 updated_at 一刷拉清。 +ai_writing: + cleanup: + enabled: true + retention_days: 7 # 保留 7 天以内的会话，更早的整条删 + cron: "0 3 * * *" # 每天 03:00 本地时区 + delete_only_finished: false # false=不看状态一刷拉；true=只清 done/error + batch_size: 500 # 单批最多删多少行，避免长事务 # 记忆子系统:跨会话持久化用户画像与 agent 私有记忆。 # 三层覆盖优先级(高->低): # users/{user_id}/memory_config.json > .deer-flow/memory_user_overrides.json > 此处