diff --git a/frontend-web/src/components/page-sidebar.tsx b/frontend-web/src/components/page-sidebar.tsx index 3c39dc32..a00206e9 100644 --- a/frontend-web/src/components/page-sidebar.tsx +++ b/frontend-web/src/components/page-sidebar.tsx @@ -1,45 +1,64 @@ -import { useState } from "react"; +import { useEffect, useState } from "react"; import type { ElementType } from "react"; import { useNavigate, useLocation } from "react-router-dom"; import { BarChart3, + Bell, Bot, Brain, + CalendarClock, ChevronLeft, ChevronRight, + ChevronsUpDown, CircleHelp, Compass, Database, FileBox, - FileOutput, FileText, Folder, FolderOpen, + Gauge, Grid3X3, HelpCircle, History, - Layers, Layout, Library, + ListChecks, ListTodo, - MessageSquare, + MessageSquareCode, + MessageSquarePlus, MoreHorizontal, Network, - PlayCircle, Plus, Scale, Settings, + Sparkles, + Tags, Target, TrendingUp, + User, Users, - Workflow, + Wrench, } from "lucide-react"; +import { + DropdownMenu, + DropdownMenuContent, + DropdownMenuItem, + DropdownMenuTrigger, +} from "@/components/ui/dropdown-menu"; +import { NotificationDialog } from "@/components/workspace/notification-dialog"; +import { SettingsDialog } from "@/components/workspace/settings"; +import { useTheme } from "@/contexts/ThemeContext"; +import { getStoredAuth } from "@/core/auth"; +import { useI18n } from "@/core/i18n/hooks"; +import { useUnreadCount } from "@/core/notifications"; type MenuItem = { id: string; label: string; icon: ElementType; path?: string; + adminOnly?: boolean; children?: MenuItem[]; }; @@ -48,19 +67,18 @@ type PageSidebarProps = { onToggleCollapsed: () => void; }; +type SettingsSection = "account" | "appearance" | "memory" | "tools" | "about"; + const menuItems: MenuItem[] = [ { id: "qa-center", label: "问答中心", icon: HelpCircle, children: [ - { id: "new-qa", label: "新建问答", icon: Plus }, - { id: "general-qa", label: "通用问答", icon: CircleHelp, path: "/page/strategy/qa/general" }, - { id: "agent-qa", label: "智能体问答", icon: Bot }, - { id: "multi-agent-qa", label: "多智能体", icon: Layers }, - { id: "sixstep-chat", label: "六步对话", icon: Workflow }, - { id: "qa-records", label: "问答记录", icon: History }, - { id: "template-editor", label: "模板配置", icon: Settings }, + { id: "general-qa", label: "通用问答", icon: CircleHelp, path: "/page/workspace/chats/new" }, + { id: "qa-zzjc", label: "zzjc支持", icon: Bot }, + { id: "qa-zzxd", label: "zzXD应用", icon: Layout }, + { id: "qa-research", label: "课题研究", icon: FileText, path: "/page/canvas/ai-writing" }, ], }, { @@ -68,6 +86,12 @@ const menuItems: MenuItem[] = [ label: "轻应用", icon: Grid3X3, children: [ + { + id: "light-app-new-note", + label: "新建笔记", + icon: Plus, + path: "/page/strategy/light-app/new-note", + }, { id: "light-app-square", label: "应用广场", icon: Layout }, { id: "light-app-records", label: "运行记录", icon: History }, ], @@ -77,10 +101,9 @@ const menuItems: MenuItem[] = [ label: "任务管理", icon: Target, children: [ - { id: "dashboard", label: "数据看板", icon: BarChart3 }, { - id: "xk-task", - label: "XK任务", + id: "xk-center-task", + label: "XK中任务", icon: Folder, children: [ { id: "xk-task-jc", label: "任务JC", icon: ListTodo }, @@ -88,50 +111,93 @@ const menuItems: MenuItem[] = [ ], }, { - id: "plan-task", - label: "规划任务", + id: "coh-center-task", + label: "COH中任务", icon: Folder, children: [ - { id: "plan-task-fx", label: "任务分析", icon: ListTodo }, - { id: "plan-task-action", label: "行动策略", icon: ListTodo }, + { id: "coh-task-3qfx", label: "3QFX", icon: ListTodo }, + { id: "coh-task-fx", label: "任务FX", icon: ListTodo }, + { id: "coh-task-xdfx", label: "XDFX", icon: ListTodo }, ], }, { - id: "mnty-task", - label: "MNTY任务", + id: "pg-center-task", + label: "PG中任务", icon: Folder, children: [ - { id: "mnty-task-mechanism", label: "机制研究", icon: ListTodo }, - { id: "mnty-task-plan", label: "规划任务", icon: ListTodo }, - { id: "mnty-task-zsty", label: "ZSTY任务", icon: ListTodo }, + { id: "pg-task-pg", label: "任务PG", icon: ListTodo }, + { id: "pg-task-xdpg", label: "XDPG", icon: ListTodo }, + { id: "pg-task-info-pg", label: "信息PG", icon: ListTodo }, ], }, { - id: "effect-task", - label: "效果PG任务", + id: "ty-center-task", + label: "TY中任务", icon: Folder, children: [ - { id: "effect-task-pg", label: "任务PG", icon: ListTodo }, - { id: "effect-action-pg", label: "行动PG", icon: ListTodo }, - { id: "effect-info-pg", label: "信息PG", icon: ListTodo }, - { id: "effect-account-pg", label: "账号PG", icon: ListTodo }, + { id: "ty-task-xdty", label: "XDTY任务", icon: ListTodo }, + { id: "ty-task-plan", label: "方案TY任务", icon: ListTodo }, + { id: "ty-task-xxty", label: "XXTY任务", icon: ListTodo }, ], }, - { id: "task-create", label: "创建任务", icon: Plus }, - { id: "task-single", label: "单智能体任务", icon: Bot }, - { id: "task-scene", label: "场景任务", icon: Workflow }, - { id: "task-consult-create", label: "多智能体协商", icon: MessageSquare }, - { id: "final-plan", label: "最终输出", icon: FileOutput }, ], }, - { id: "scene-list", label: "场景管理", icon: Layers }, { - id: "agent-list", - label: "智能体管理", - icon: Bot, + id: "comprehensive-management", + label: "综合管理", + icon: Settings, children: [ - { id: "agent-overview", label: "智能体列表", icon: Users, path: "/page/strategy/agents" }, - { id: "agent-trial", label: "单智能体试用", icon: PlayCircle }, + { id: "mgmt-agents", label: "智能体管理", icon: Bot, path: "/page/strategy/agents" }, + { id: "mgmt-skills", label: "技能管理", icon: Sparkles, path: "/page/strategy/skills" }, + { id: "mgmt-memory", label: "记忆管理", icon: Brain, path: "/page/strategy/memory" }, + { + id: "mgmt-tasks", + label: "任务管理", + icon: ListChecks, + path: "/page/strategy/scheduled/tasks", + }, + { + id: "mgmt-recommended-questions", + label: "推荐问题配置", + icon: MessageSquarePlus, + path: "/page/strategy/settings/recommended-questions", + adminOnly: true, + }, + { + id: "mgmt-prompt-prefix", + label: "问答前缀配置", + icon: MessageSquareCode, + path: "/page/strategy/settings/prompt-prefix", + adminOnly: true, + }, + { + id: "mgmt-llm-metrics", + label: "大模型调用统计", + icon: Gauge, + path: "/page/strategy/settings/llm-metrics", + adminOnly: true, + }, + { + id: "mgmt-users", + label: "用户管理", + icon: Users, + path: "/page/strategy/admin/users", + adminOnly: true, + }, + { + id: "mgmt-admin-scheduled-tasks", + label: "定时任务管理", + icon: CalendarClock, + path: "/page/strategy/admin/scheduled-tasks", + adminOnly: true, + }, + { + id: "mgmt-user-activity", + label: "用户活动审计", + icon: History, + path: "/page/strategy/admin/user-activity", + adminOnly: true, + }, ], }, { @@ -214,6 +280,13 @@ const menuItems: MenuItem[] = [ label: "资源管理", icon: Database, children: [ + { + id: "resource-tags", + label: "标签管理", + icon: Tags, + path: "/page/strategy/admin/tags", + adminOnly: true, + }, { id: "resource-zz", label: "ZZ资源", icon: FileBox }, { id: "resource-network-actor", label: "网络达人", icon: Network }, { id: "resource-knowledge-base", label: "知识库", icon: Library }, @@ -225,26 +298,72 @@ const menuItems: MenuItem[] = [ const defaultExpandedIds = new Set([ "qa-center", "task-management", - "xk-task", + "xk-center-task", "result-management", "result-planning", ]); +function filterMenuItemsByRole(items: MenuItem[], isAdmin: boolean): MenuItem[] { + return items + .filter((item) => !item.adminOnly || isAdmin) + .map((item) => + item.children + ? { ...item, children: filterMenuItemsByRole(item.children, isAdmin) } + : item, + ); +} + export function PageSidebar({ collapsed, onToggleCollapsed, }: PageSidebarProps) { + const { theme } = useTheme(); + const isDark = theme === "dark"; + const isAdmin = getStoredAuth()?.system_role === "admin"; + const visibleMenuItems = filterMenuItemsByRole(menuItems, !!isAdmin); + const { t } = useI18n(); + const { data: unreadCount = 0 } = useUnreadCount(); const [expandedIds, setExpandedIds] = useState>(defaultExpandedIds); + const [settingsOpen, setSettingsOpen] = useState(false); + const [settingsDefaultSection, setSettingsDefaultSection] = + useState("appearance"); + const [notificationOpen, setNotificationOpen] = useState(false); + const [mounted, setMounted] = useState(false); const navigate = useNavigate(); const { pathname } = useLocation(); - const iconButtonClass = - "rounded-lg p-2 text-gray-500 transition-all duration-300 ease-out hover:scale-105 hover:bg-red-50 hover:text-red-600"; - const panelClass = "ml-4 space-y-1 border-l border-red-100 pl-4"; + useEffect(() => { + setMounted(true); + }, []); + + const openSettings = (section: SettingsSection) => { + setSettingsDefaultSection(section); + setSettingsOpen(true); + }; + + const iconButtonClass = isDark + ? "rounded-lg p-2 text-slate-400 transition-all duration-300 ease-out hover:scale-105 hover:bg-slate-800/50 hover:text-white" + : "rounded-lg p-2 text-gray-500 transition-all duration-300 ease-out hover:scale-105 hover:bg-red-50 hover:text-red-600"; + const panelClass = isDark + ? "ml-4 space-y-1 border-l border-blue-800/30 pl-4" + : "ml-4 space-y-1 border-l border-red-100 pl-4"; const labelClass = (compact: boolean, small = false) => `overflow-hidden whitespace-nowrap transition-all duration-300 ease-out ${ compact ? "max-w-0 opacity-0" : "max-w-[220px] opacity-100" } ${small ? "text-xs" : "text-sm"} flex-1 font-medium`; + const menuButtonClass = (isActive: boolean, depth = 0) => + `flex w-full items-center rounded-lg py-2 text-left transition-all duration-300 ease-out hover:-translate-y-0.5 theme-transition ${ + collapsed && depth === 0 ? "justify-center px-2" : "gap-2 px-3" + } ${ + isActive + ? isDark + ? "bg-blue-500/15 text-cyan-300 font-semibold" + : "bg-red-50 text-red-600 font-semibold" + : isDark + ? "text-slate-400 hover:bg-slate-800/50 hover:text-white" + : "text-gray-600 hover:bg-red-50 hover:text-red-600" + }`; + const footerButtonClass = menuButtonClass(false, 0); const toggleExpanded = (id: string) => { setExpandedIds((prev) => { @@ -268,13 +387,7 @@ export function PageSidebar({ type="button" title={collapsed && depth === 0 ? item.label : undefined} onClick={() => item.path && navigate(item.path)} - className={`flex w-full items-center rounded-lg py-2 text-left transition-all duration-300 ease-out hover:-translate-y-0.5 ${ - collapsed && depth === 0 ? "justify-center px-2" : "gap-2 px-3" - } ${ - isActive - ? "bg-red-50 text-red-600 font-semibold" - : "text-gray-600 hover:bg-red-50 hover:text-red-600" - }`} + className={menuButtonClass(isActive, depth)} > {(!collapsed || depth > 0) && ( @@ -303,9 +416,7 @@ export function PageSidebar({ toggleExpanded(item.id); }} title={collapsed && isTopLevel ? item.label : undefined} - className={`flex w-full items-center rounded-lg py-2 text-left text-gray-600 transition-all duration-300 ease-out hover:-translate-y-0.5 hover:bg-red-50 hover:text-red-600 ${ - collapsed && isTopLevel ? "justify-center px-2" : "gap-2 px-3" - }`} + className={menuButtonClass(false, depth)} > + <> + + + + ); } diff --git a/frontend-web/src/components/workspace/workspace-header.tsx b/frontend-web/src/components/workspace/workspace-header.tsx index 11eb5b31..ba1767d8 100644 --- a/frontend-web/src/components/workspace/workspace-header.tsx +++ b/frontend-web/src/components/workspace/workspace-header.tsx @@ -23,7 +23,7 @@ function BrandMark({ className }: { className?: string }) { ); } -function SidebarCollapseButton({ className }: { className?: string }) { +export function SidebarCollapseButton({ className }: { className?: string }) { const { open, toggleSidebar } = useSidebar(); return ( + )} + + + {/* 说明文字 */} +
+ +

+ 新建笔记开始使用 +

+
+ + + ); +} diff --git a/frontend-web/src/pages/LoginPage.tsx b/frontend-web/src/pages/LoginPage.tsx index 559b6a83..16ee95b9 100644 --- a/frontend-web/src/pages/LoginPage.tsx +++ b/frontend-web/src/pages/LoginPage.tsx @@ -1,10 +1,14 @@ "use client"; -import { useParams, useRouter } from "next/navigation"; +import { useParams, useRouter, useSearchParams } from "next/navigation"; import { useEffect, useMemo, useState } from "react"; import { loginByUsername } from "@/core/auth/api"; import { setStoredAuth } from "@/core/auth"; +import { + isSidebarEnabledFromSearch, + persistSidebarEnabled, +} from "@/core/page-layout/sidebar-enabled"; const DEFAULT_USERNAME = "guest"; @@ -13,6 +17,11 @@ type LoginStatus = "loading" | "success" | "error"; export default function LoginPage() { const { username } = useParams<{ username?: string }>(); const router = useRouter(); + const searchParams = useSearchParams(); + const sidebarEnabled = useMemo( + () => isSidebarEnabledFromSearch(searchParams.toString()), + [searchParams], + ); const resolvedUsername = useMemo(() => { const raw = username?.trim(); return raw && raw.length > 0 ? raw : DEFAULT_USERNAME; @@ -32,6 +41,7 @@ export default function LoginPage() { return; } setStoredAuth(auth); + persistSidebarEnabled(sidebarEnabled); setStatus("success"); router.replace("/page/workspace/chats/new"); } catch (error) { @@ -49,7 +59,7 @@ export default function LoginPage() { return () => { cancelled = true; }; - }, [resolvedUsername, router]); + }, [resolvedUsername, router, sidebarEnabled]); return (
diff --git a/frontend-web/src/pages/PageRoutes.tsx b/frontend-web/src/pages/PageRoutes.tsx index 4c7d6fa8..4ae216da 100644 --- a/frontend-web/src/pages/PageRoutes.tsx +++ b/frontend-web/src/pages/PageRoutes.tsx @@ -1,22 +1,21 @@ -import { useState } from "react"; +import { useMemo, useState } from "react"; import { Navigate, Route, Routes } from "react-router-dom"; import { PageSidebar } from "@/components/page-sidebar"; +import { readSidebarEnabled } from "@/core/page-layout/sidebar-enabled"; import WorkspaceRoutes from "./WorkspaceRoutes"; import StrategyRoutes from "./StrategyRoutes"; import Header from "@/strategy-components/Header"; import OpenCanvasRoutes from "@/open-canvas/routes/OpenCanvasRoutes"; -// 设为 true 时隐藏侧边栏（开发中，暂未完成） -const SIDEBAR_ENABLED = false; - function PageLayout() { const [collapsed, setCollapsed] = useState(false); + const sidebarEnabled = useMemo(() => readSidebarEnabled(), []); return (
- {SIDEBAR_ENABLED && ( + {sidebarEnabled && ( setCollapsed((prev) => !prev)} @@ -24,10 +23,10 @@ function PageLayout() { )}
- {SIDEBAR_ENABLED &&
} + {sidebarEnabled &&
}
} /> @@ -36,8 +35,9 @@ function PageLayout() { element={ } /> diff --git a/frontend-web/src/pages/StrategyRoutes.tsx b/frontend-web/src/pages/StrategyRoutes.tsx index 42eadc32..aa1b8039 100644 --- a/frontend-web/src/pages/StrategyRoutes.tsx +++ b/frontend-web/src/pages/StrategyRoutes.tsx @@ -12,9 +12,19 @@ import ChatsPage from "./ChatsPage"; import ChatPage from "./ChatPage"; import NewAgentPage from "./NewAgentPage"; import AgentChatPage from "./AgentChatPage"; +import SkillsPage from "./SkillsPage"; +import SkillEvolutionPage from "./SkillEvolutionPage"; +import MemoryPage from "./MemoryPage"; import ScheduledTasksPage from "./ScheduledTasksPage"; import ScheduledTaskRunDetailPage from "./ScheduledTaskRunDetailPage"; import RecommendedQuestionsPage from "./RecommendedQuestionsPage"; +import PromptPrefixSettingsPage from "./PromptPrefixSettingsPage"; +import LlmMetricsPage from "./LlmMetricsPage"; +import UsersPage from "./UsersPage"; +import AdminScheduledTasksPage from "./AdminScheduledTasksPage"; +import AdminUserActivityPage from "./AdminUserActivityPage"; +import TagsPage from "./TagsPage"; +import LightAppNewNotePage from "./LightAppNewNotePage"; export default function StrategyRoutes() { return ( @@ -27,6 +37,9 @@ export default function StrategyRoutes() { {/* 问答中心 */} } /> + {/* 轻应用 */} + } /> + {/* 对话 */} } /> + {/* 技能 / 记忆 */} + } /> + } /> + } /> + {/* 定时任务 */} } /> - {/* 设置 */} + {/* 设置 / 管理 — 权限与 workspace 侧一致，非 admin 重定向到 chats/new */} ) : ( - + + ) + } + /> + + ) : ( + + ) + } + /> + + ) : ( + + ) + } + /> + + ) : ( + + ) + } + /> + + ) : ( + + ) + } + /> + + ) : ( + + ) + } + /> + + ) : ( + ) } /> diff --git a/frontend-web/src/pages/WorkspaceRoutes.tsx b/frontend-web/src/pages/WorkspaceRoutes.tsx index c8ad9871..2471e375 100644 --- a/frontend-web/src/pages/WorkspaceRoutes.tsx +++ b/frontend-web/src/pages/WorkspaceRoutes.tsx @@ -37,11 +37,13 @@ export function WorkspaceLayout({ pageSidebarCollapsed, pageSidebarHidden = false, headerHidden = false, + compactWorkspaceSidebar = false, }: { children: ReactNode; pageSidebarCollapsed: boolean; pageSidebarHidden?: boolean; headerHidden?: boolean; + compactWorkspaceSidebar?: boolean; }) { const pageSidebarWidth = pageSidebarHidden ? "0rem" @@ -63,7 +65,7 @@ export function WorkspaceLayout({ data-page-sidebar-collapsed={pageSidebarCollapsed} defaultOpen > - + {children} @@ -85,12 +87,19 @@ export default function WorkspaceRoutes({ pageSidebarCollapsed = false, pageSidebarHidden = false, headerHidden = false, + compactWorkspaceSidebar = false, }: { pageSidebarCollapsed?: boolean; pageSidebarHidden?: boolean; headerHidden?: boolean; + compactWorkspaceSidebar?: boolean; }) { - const layoutProps = { pageSidebarCollapsed, pageSidebarHidden, headerHidden }; + const layoutProps = { + pageSidebarCollapsed, + pageSidebarHidden, + headerHidden, + compactWorkspaceSidebar, + }; return ( diff --git a/frontend-web/src/strategy-components/Header.tsx b/frontend-web/src/strategy-components/Header.tsx index f5b1c716..c446a297 100644 --- a/frontend-web/src/strategy-components/Header.tsx +++ b/frontend-web/src/strategy-components/Header.tsx @@ -1,31 +1,29 @@ -import React from 'react'; +import React, { useState } from 'react'; import { Bell, Search, User, ChevronDown, Moon, Palette } from 'lucide-react'; +import { NotificationDialog } from '@/components/workspace/notification-dialog'; import { useTheme } from '../contexts/ThemeContext'; +import { useUnreadCount } from '@/core/notifications'; const Header: React.FC = () => { const { theme, toggleTheme } = useTheme(); const isDark = theme === 'dark'; + const [notificationOpen, setNotificationOpen] = useState(false); + const { data: unreadCount = 0 } = useUnreadCount(); return ( -
+ +
{/* 左侧 - 搜索框 */}
-
- - -
+
{/* 右侧 - 用户信息和通知 */} @@ -44,11 +42,24 @@ const Header: React.FC = () => { {/* 通知 */} - {/* 分隔线 */} @@ -71,6 +82,7 @@ const Header: React.FC = () => {
+ ); }; diff --git a/offline-backend-20260512/backend/app/gateway/ai_writing_cleanup.py b/offline-backend-20260512/backend/app/gateway/ai_writing_cleanup.py index 78673a7b..0399cb00 100644 --- a/offline-backend-20260512/backend/app/gateway/ai_writing_cleanup.py +++ b/offline-backend-20260512/backend/app/gateway/ai_writing_cleanup.py @@ -108,25 +108,31 @@ async def _delete_checkpoint_thread(checkpointer, thread_id: str) -> None: ) -def _get_active_session_ids() -> set[str]: - """返回内存里**当前正在跑**的 session_id 集合。 +async def _get_active_session_ids_langgraph(thread_store) -> set[str]: + """新路径下「正在跑 / 等用户干预」的 thread_id 集合。 - 这些会话的写作流程还活着 —— 任何清理都不能动它们，否则用户「正在写」 - 的稿子会被秒删，体验灾难。延迟到下一轮清理（24 小时后）再处理也比误 - 删强。 + ``threads_meta.status`` 由 ``start_run`` / ``run_agent`` 维护： + - ``running``：graph 正在跑（节点中） + - ``interrupted``：graph 暂停在 interrupt 点等用户回头填卡 + 这两种都视作活跃，**绝不能删**。``idle / success / error`` 才是真正终止状态。 + + LangGraph Server 路径 + 老路径的 lead_agent 都用同一份 ``thread_store``，所以 + 这里拿的 active 集合是全部 graph 的；筛 ai_writing 是按 assistant_id 过滤。 """ - 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) + if thread_store is None: return set() + try: + running = await thread_store.search(status="running", user_id=None, limit=1000) + interrupted = await thread_store.search(status="interrupted", user_id=None, limit=1000) + except Exception: + logger.warning("ai_writing_cleanup: 读取 LangGraph 活动 thread 列表失败", exc_info=True) + return set() + # 仅关心 assistant_id='ai_writing' 的 thread —— 别的 graph 不归这套清理管 + active: set[str] = set() + for row in running + interrupted: + if row.get("assistant_id") == "ai_writing": + active.add(row["thread_id"]) + return active async def run_cleanup_once( @@ -134,18 +140,27 @@ async def run_cleanup_once( retention_days: int, only_finished: bool = False, batch_size: int = 500, + checkpointer=None, + thread_store=None, ) -> dict: """执行一轮清理。返回统计信息。 - **安全保证**：清理永远跳过内存里仍在跑的会话（``_event_queues`` 里有 - 对应 key 的）—— 这条规则即使在 admin 手动触发清理时也强制生效，防止 - 把用户「正在写」的稿子误删。 + Args: + checkpointer: 主 LangGraph checkpointer（``app.state.checkpointer``）。 + AI 写作的 thread 状态由 LangGraph Server 注入，存在这里 —— admin 路由 + + scheduler loop 都会从 ``app.state`` 取到引用传进来；测试场景下传 None + 会让 checkpoint 删除变成 no-op，但 DB 行仍会被清。 + thread_store: ``app.state.thread_store``。用来按 ``status in (running, + interrupted)`` 查活跃 thread 列表做活跃保护。 + + **安全保证**：清理永远跳过仍在跑 / 等用户干预的会话（``threads_meta.status + in (running, interrupted)`` + ``assistant_id='ai_writing'``）。即使 admin + 手动触发 ``retention_days=0`` 也强制生效。 幂等 + 可重入 —— 多次调用对结果只会越扫越干净。即便清理途中被中断，下次 再跑会重新把 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 @@ -155,8 +170,7 @@ async def run_cleanup_once( return {"deleted_db": 0, "deleted_checkpoints": 0, "skipped_active": 0} repo = AIWritingSessionRepository(sf) - checkpointer = get_ai_writing_checkpointer() - active_ids = _get_active_session_ids() + active_ids = await _get_active_session_ids_langgraph(thread_store) cutoff = datetime.now(UTC) - timedelta(days=retention_days) logger.info( @@ -238,11 +252,16 @@ def _next_cron_delay_seconds(cron_expr: str, base: datetime | None = None) -> fl return max(1.0, (nxt - base).total_seconds()) -async def cleanup_scheduler_loop() -> None: +async def cleanup_scheduler_loop(app=None) -> None: """常驻后台任务：按 cron 调度跑清理。 多 worker 部署下文件锁做 leader election，没拿到锁的 worker 直接退出。 被 cancel 时 CancelledError 不会被 except Exception 吞掉，能干净退出。 + + Args: + app: FastAPI 实例，用来从 ``app.state`` 拿主 checkpointer / thread_store + 引用，喂给 ``run_cleanup_once`` —— 新路径下 ai_writing 的 thread 状态 + 在主 checkpointer 里。``app=None`` 时回退到老 checkpointer（仅老路径）。 """ from deerflow.config.app_config import get_app_config from deerflow.config.runtime_paths import runtime_home @@ -289,10 +308,17 @@ async def cleanup_scheduler_loop() -> None: logger.info("ai_writing_cleanup: 下一次清理 %.0f 秒后", delay) await asyncio.sleep(delay) try: + # 每次循环重新读 app.state —— 不缓存，避免 lifespan 中重启 / 重置 + # checkpointer 时拿到死引用。app=None 时 main_checkpointer / store + # 也为 None，run_cleanup_once 内部走老 fallback。 + main_checkpointer = getattr(getattr(app, "state", None), "checkpointer", None) + main_thread_store = getattr(getattr(app, "state", None), "thread_store", None) await run_cleanup_once( retention_days=retention_days, only_finished=only_finished, batch_size=batch_size, + checkpointer=main_checkpointer, + thread_store=main_thread_store, ) except Exception: logger.exception("ai_writing_cleanup: 本轮清理失败（不影响下一轮）") diff --git a/offline-backend-20260512/backend/app/gateway/app.py b/offline-backend-20260512/backend/app/gateway/app.py index 2e161eb2..30152129 100644 --- a/offline-backend-20260512/backend/app/gateway/app.py +++ b/offline-backend-20260512/backend/app/gateway/app.py @@ -71,12 +71,6 @@ 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. @@ -371,11 +365,10 @@ async def _curator_scheduler_loop() -> None: 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 走不通。 + 主 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），不应该阻止启动。 @@ -474,7 +467,7 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]: try: from app.gateway.ai_writing_cleanup import cleanup_scheduler_loop - app.state.ai_writing_cleanup_task = asyncio.create_task(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)") @@ -526,32 +519,9 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]: 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") + # 阶段 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) diff --git a/offline-backend-20260512/backend/app/gateway/deps.py b/offline-backend-20260512/backend/app/gateway/deps.py index 2d630d1b..3eff6b6e 100644 --- a/offline-backend-20260512/backend/app/gateway/deps.py +++ b/offline-backend-20260512/backend/app/gateway/deps.py @@ -130,9 +130,15 @@ async def langgraph_runtime(app: FastAPI) -> AsyncGenerator[None, None]: app.state.thread_share_store = make_thread_share_store(sf) - from deerflow.persistence.ai_writing_sessions import make_ai_writing_session_store + from deerflow.persistence.ai_writing_sessions import ( + make_ai_writing_session_store, + register_default_store as _register_ai_writing_default_store, + ) app.state.ai_writing_session_store = make_ai_writing_session_store(sf) + # 让 LangGraph Server 路径下的 AI 写作节点能跨 app/harness 边界拿到 repo。 + # 老 /start /stream /resume 路由仍走 request.app.state，互不影响。 + _register_ai_writing_default_store(app.state.ai_writing_session_store) from deerflow.persistence.article_types import make_article_type_store 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 968c6408..70fe1105 100644 --- a/offline-backend-20260512/backend/app/gateway/routers/ai_writing.py +++ b/offline-backend-20260512/backend/app/gateway/routers/ai_writing.py @@ -1,31 +1,26 @@ -"""AI 写作路由 —— 独立于原有路由，前缀 /api/ai-writing/ +"""AI 写作路由 —— 业务表 CRUD + transcript 持久化 + admin cleanup。 + +写作主流程（``/start /stream /resume /state``）已迁到 LangGraph Server 的 +``ai_writing`` assistant（前端走 ``useAIWritingStream`` hook + SDK），路由层 +不再承载 graph 推进，只保留以下「不依赖 graph 状态」的辅助接口： -POST /api/ai-writing/start 启动写作流程，返回 session_id -GET /api/ai-writing/{id}/stream SSE 流（进度/干预/草稿事件） -POST /api/ai-writing/{id}/resume 用户干预响应，恢复流程 -GET /api/ai-writing/{id}/state 查询当前流程状态 GET /api/ai-writing/sessions 列出用户的会话历史 GET /api/ai-writing/sessions/{id} 获取单个历史会话 PATCH /api/ai-writing/sessions/{id} 重命名会话标题 DELETE /api/ai-writing/sessions/{id} 删除会话 +PUT /api/ai-writing/sessions/{id}/transcript 保存完整对话时间线 GET /api/ai-writing/article-types 列出所有文章类型 POST /api/ai-writing/article-types 新增文章类型 PATCH /api/ai-writing/article-types/{id} 更新文章类型 DELETE /api/ai-writing/article-types/{id} 删除文章类型 +POST /api/ai-writing/admin/cleanup （admin）手动清理会话 """ from __future__ import annotations -import asyncio -import json import logging -import uuid -from collections.abc import AsyncGenerator -from datetime import UTC, datetime from typing import Optional from fastapi import APIRouter, HTTPException, Request -from fastapi.responses import StreamingResponse -from langgraph.types import Command from pydantic import BaseModel, Field logger = logging.getLogger(__name__) @@ -34,38 +29,6 @@ router = APIRouter(prefix="/api/ai-writing", tags=["ai-writing"]) # ── 请求/响应模型 ───────────────────────────────────────────────────────────── -class StartWritingRequest(BaseModel): - user_intent: str - article_type: str = "analysis" - target_audience: str = "普通读者" - word_count_target: int = 800 - model_name: str = "" - # 素材检索关键词数量：用户指定用几个关键词检索（1~10，默认 6） - keyword_count: int = 6 - # 写作模式：strict=严格依据素材 / loose=可结合通用知识（默认 loose） - writing_mode: str = "loose" - - -class StartWritingResponse(BaseModel): - session_id: str - message: str = "写作流程已启动" - - -class ResumeRequest(BaseModel): - action: str - pause_point: str = "" - approved_material_ids: list[str] = Field(default_factory=list) - extra_keywords: list[str] = Field(default_factory=list) - edited_outline: Optional[dict] = None - outline_feedback: str = "" - user_revision_notes: str = "" - override_verdict: bool = False - selected_issue_indices: Optional[list[int]] = None - # 素材不足求助点：每个被阻塞章节的处理决策 - # [{"sectionTitle": "...", "decision": "supplement|delete|loose"}] - section_decisions: list[dict] = Field(default_factory=list) - - class RenameSessionRequest(BaseModel): title: str @@ -85,8 +48,8 @@ class AdminCleanupRequest(BaseModel): class AdminCleanupResponse(BaseModel): deleted_db: int deleted_checkpoints: int - # 因「仍在跑」被保护掉的会话数。永远跳过 _event_queues 里活跃的会话， - # 即使 retention_days=0 也不会秒杀用户正在写的稿。 + # 因「仍在跑 / 等用户干预」被保护掉的会话数（threads_meta.status in + # running/interrupted）。即使 retention_days=0 也不会秒杀用户正在写的稿。 skipped_active: int = 0 elapsed_seconds: float params: dict @@ -142,244 +105,7 @@ class SessionResponse(BaseModel): updated_at: str -# ── 内部状态 ────────────────────────────────────────────────────────────────── - -# session_id → asyncio.Queue -_event_queues: dict[str, asyncio.Queue] = {} - -# session_id → model_name（resume 时复用） -_session_model_names: dict[str, str] = {} - -# session_id → AIWritingSessionRepository（for DB saves） -_session_repos: dict[str, object] = {} - -# session_id → user_id (str | None, already resolved) -_session_user_ids: dict[str, str | None] = {} - -# session_id → accumulated intervention records -_session_interventions: dict[str, list] = {} - -# session_id → True if session ended with error -_session_error_flags: dict[str, bool] = {} - -# session_id → valid_article_types（list[dict] with key/label） -_session_article_types: dict[str, list[dict]] = {} - -# session_id → pending await_user event tuple for SSE reconnect re-delivery -# asyncio.Queue is consume-once; if the old SSE generator dequeued the await_user event -# before the client disconnected, the event is lost from the queue. We save a copy here -# so that a new SSE connection (e.g., after user clicks "继续") can re-inject it. -_session_pending_pause: dict[str, tuple] = {} - -# Serializes session recovery so that a concurrent /stream + /resume on the same -# 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: - """组装一帧 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: - """读取 config.yaml 的 ai_writing 段（含素材检索 provider 配置）。""" - try: - from deerflow.config import get_app_config - - aw = getattr(get_app_config(), "ai_writing", None) - if isinstance(aw, dict): - return aw - return dict(aw) if aw else {} - except Exception: - return {} - - -def _make_config(session_id: str, model_name: str = "", valid_article_types: list[dict] | None = None) -> dict: - return { - "configurable": { - "thread_id": session_id, - "model_name": model_name or _session_model_names.get(session_id, ""), - "ai_writing_config": _ai_writing_config(), - # 把 SSE 队列注入，供草稿/大纲节点直接推流式 chunk - "sse_queue": _event_queues.get(session_id), - "sse_session_id": session_id, - # 动态文章类型列表，intent_parser 从此处读取有效枚举 - "valid_article_types": valid_article_types or _session_article_types.get(session_id), - } - } - - -def _truncate_title(text: str, max_len: int = 50) -> str: - text = text.strip() - 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。 - - **失败兜底契约**：本接口除非 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 - _session_model_names[session_id] = body.model_name - _session_interventions[session_id] = [] - - # 加载动态文章类型列表，注入给 intent_parser 节点 - article_type_store = get_article_type_store(request) - valid_article_types: list[dict] | None = None - if article_type_store is not None: - try: - records = await article_type_store.list() - if records: - valid_article_types = [{"key": r["key"], "label": r["label"]} for r in records] - except Exception: - logger.exception("ai_writing: failed to load article types for session=%s", session_id) - _session_article_types[session_id] = valid_article_types or [] - - # 持久化会话记录（DB 可能未配置，则跳过） - repo = get_ai_writing_session_store(request) - user_id = await get_current_user(request) - if repo is not None: - try: - title = _truncate_title(body.user_intent) - await repo.create( - id=session_id, - title=title, - user_intent=body.user_intent, - user_id=user_id, - ) - _session_repos[session_id] = repo - _session_user_ids[session_id] = user_id - except Exception: - logger.exception("ai_writing: failed to create session record: session=%s", session_id) - - initial_state = { - "user_intent": body.user_intent, - "article_type": body.article_type, - "target_audience": body.target_audience, - "word_count_target": body.word_count_target, - # 规整到 [1, 10]，非法值回退默认 6 - "keyword_count": max(1, min(10, body.keyword_count)), - # 写作模式，非法值回退 loose - "writing_mode": body.writing_mode if body.writing_mode in ("strict", "loose") else "loose", - "revision_count": 0, - "max_revisions": 3, - "status": "researching", - "drafts": [], - "progress_events": [], - "material_package": None, - "current_outline": None, - "current_draft": None, - "review_result": None, - "last_user_intervention": None, - } - config = _make_config(session_id, body.model_name, valid_article_types=valid_article_types) - - _spawn_graph_task(_stream_graph_segment(session_id, initial_state, config, graph)) - return StartWritingResponse(session_id=session_id) - - @router.get("/sessions", response_model=list[SessionResponse]) async def list_sessions(request: Request): """列出当前用户的 AI 写作历史会话。""" @@ -460,8 +186,6 @@ async def save_transcript(session_id: str, body: SaveTranscriptRequest, request: # 会话不存在或不属于当前用户：同样不抛 500，返回 success=False 即可 return {"success": False, "reason": "session_not_found"} return {"success": True} - - @router.get("/article-types", response_model=list[ArticleTypeResponse]) async def list_article_types(request: Request): """列出所有文章类型，按 sort_order 升序排列。""" @@ -535,91 +259,6 @@ async def delete_article_type(article_type_id: str, request: Request): return {"success": True} -@router.get("/{session_id}/stream") -async def stream_progress(session_id: str): - """SSE 流：推送进度事件、干预请求、草稿内容。""" - return StreamingResponse( - _sse_generator(session_id), - media_type="text/event-stream", - headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"}, - ) - - -@router.post("/{session_id}/resume") -async def resume_writing(session_id: str, body: ResumeRequest, request: Request): - """用户提交干预决策，恢复暂停的流程。""" - 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 中断整个写作流程。 - recovery = await _recover_session(session_id, request) - if recovery != "interrupted": - logger.warning( - "ai_writing: resume on unrecoverable session=%s (recovery=%s)", session_id, recovery - ) - raise HTTPException( - status_code=404, - detail="会话状态已丢失或流程已结束，无法继续，请重新发起写作。", - ) - - config = _make_config(session_id) - intervention = { - "action": body.action, - "approvedMaterialIds": body.approved_material_ids, - "extraKeywords": body.extra_keywords, - "editedOutline": body.edited_outline, - "outlineFeedback": body.outline_feedback, - "userRevisionNotes": body.user_revision_notes, - "overrideVerdict": body.override_verdict, - "selectedIssueIndices": body.selected_issue_indices, - "sectionDecisions": body.section_decisions, - } - - # Record this intervention for session history - record = { - "pausePoint": body.pause_point, - "action": body.action, - "timestamp": datetime.now(UTC).isoformat(), - } - if session_id in _session_interventions: - _session_interventions[session_id].append(record) - repo = _session_repos.get(session_id) - user_id = _session_user_ids.get(session_id) - if repo is not None: - try: - await repo.update( - session_id, - user_id=user_id, - completed_interventions=_session_interventions[session_id], - ) - except Exception: - logger.exception("ai_writing: failed to save intervention: session=%s", session_id) - - # Clear pending pause so SSE reconnect won't re-inject the stale await_user event - _session_pending_pause.pop(session_id, None) - - _spawn_graph_task( - _stream_graph_segment(session_id, Command(resume=intervention), config, graph) - ) - return {"message": "已提交，流程恢复中"} - - -@router.get("/{session_id}/state") -async def get_state(session_id: str): - """查询当前流程状态快照。""" - from deerflow.agents.ai_writing.graph import get_ai_writing_graph - - graph = await get_ai_writing_graph() - config = _make_config(session_id) - 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 - # ── 管理接口 ────────────────────────────────────────────────────────────────── @@ -688,9 +327,20 @@ async def admin_cleanup( await _require_admin(request) params = _resolve_cleanup_params(body) + # 透传主 LangGraph checkpointer / thread_store —— 新路径下 ai_writing 的 + # thread 状态在主 checkpointer 里，thread_store 用来按 status 过滤活跃 thread。 + # getattr 嵌套防御：测试用 stub Request 时 ``request.app`` 可能不存在。 + _app_state = getattr(getattr(request, "app", None), "state", None) + main_checkpointer = getattr(_app_state, "checkpointer", None) + main_thread_store = getattr(_app_state, "thread_store", None) + started = time.monotonic() try: - result = await run_cleanup_once(**params) + result = await run_cleanup_once( + **params, + checkpointer=main_checkpointer, + thread_store=main_thread_store, + ) except Exception as e: logger.exception("ai_writing admin cleanup failed: params=%s", params) # 抛 500 是合理的 —— 管理员触发清理失败需要看到失败原因，跟「业务 @@ -715,292 +365,3 @@ async def admin_cleanup( elapsed_seconds=round(elapsed, 3), params=params, ) - - -# ── 内部辅助 ────────────────────────────────────────────────────────────────── - -async def _stream_graph_segment( - session_id: str, - input_or_command, - config: dict, - graph, -): - """运行一个 Graph 片段（初始运行或 resume），将输出推入队列。 - 遇到 interrupt 时停止（不关闭队列），等待下次 resume。 - Graph 结束时关闭队列（放入 None 哨兵）。 - """ - queue = _event_queues.get(session_id) - if queue is None: - return - - reached_end = False - is_error = False - try: - async for chunk in graph.astream(input_or_command, config=config, stream_mode="updates"): - node_name = next(iter(chunk)) - node_output = chunk[node_name] - - # interrupt() 触发 - if node_name == "__interrupt__": - # node_output 是 (Interrupt(...),) 元组 - interrupt_list = node_output if isinstance(node_output, (list, tuple)) else [node_output] - for intr in interrupt_list: - interrupt_data = intr.value if hasattr(intr, "value") else intr - event_tuple = ("await_user", interrupt_data) - await queue.put(event_tuple) - # Save for SSE reconnect re-delivery: if the old generator consumed this - # event before the client disconnected, a new connection would block forever. - _session_pending_pause[session_id] = event_tuple - return # 等待 resume，不关闭队列 - - if not isinstance(node_output, dict): - continue - - # 推送 progress_events - for evt in node_output.get("progress_events") or []: - await queue.put((_event_type(evt), {**evt, "session_id": session_id})) - - # 素材就绪 - if pkg := node_output.get("material_package"): - await queue.put(("materials_ready", { - "session_id": session_id, - "keywords": pkg.get("keywords", []), - "materials": pkg.get("materials", []), - "summary": pkg.get("summary", ""), - })) - - # 大纲就绪 - if outline := node_output.get("current_outline"): - await queue.put(("outline_ready", {"session_id": session_id, "outline": outline})) - - # 草稿就绪 — 保存到 DB - if draft := node_output.get("current_draft"): - draft_title = draft.get("title", "") - draft_markdown = draft.get("full_markdown", "") - await queue.put(("draft_ready", { - "session_id": session_id, - "title": draft_title, - "markdown": draft_markdown, - "revision_count": draft.get("revision_count", 0), - })) - repo = _session_repos.get(session_id) - user_id = _session_user_ids.get(session_id) - if repo is not None: - try: - await repo.update( - session_id, - user_id=user_id, - status="draft_ready", - title=draft_title or _truncate_title( - next(iter(_session_interventions.get(session_id, [{}])), {}).get("action", "草稿"), - ), - draft_title=draft_title, - draft_markdown=draft_markdown, - ) - except Exception: - logger.exception("ai_writing: failed to save draft: session=%s", session_id) - - # 编辑审核就绪 — 保存到 DB - if review := node_output.get("review_result"): - await queue.put(("review_ready", {"session_id": session_id, "review_result": review})) - repo = _session_repos.get(session_id) - user_id = _session_user_ids.get(session_id) - if repo is not None: - try: - await repo.update( - session_id, - user_id=user_id, - status="review_ready", - review_result=review, - ) - except Exception: - logger.exception("ai_writing: failed to save review: session=%s", session_id) - - # 状态 done：Graph 结束 - if node_output.get("status") == "done": - reached_end = True - - # astream 正常结束 = Graph 执行到 END - reached_end = True - await queue.put(("done", {"session_id": session_id, "message": "写作完成"})) - - except Exception as e: - logger.exception("ai_writing graph segment failed: session=%s", session_id) - await queue.put(("error", {"session_id": session_id, "message": str(e)})) - is_error = True - reached_end = True - finally: - if reached_end: - # 更新最终状态到 DB - repo = _session_repos.get(session_id) - user_id = _session_user_ids.get(session_id) - if repo is not None: - try: - await repo.update( - session_id, - user_id=user_id, - status="error" if is_error else "done", - ) - except Exception: - logger.exception("ai_writing: failed to update final status: session=%s", session_id) - - await queue.put(None) # 哨兵，通知 SSE 生成器关闭 - _event_queues.pop(session_id, None) - _session_model_names.pop(session_id, None) - _session_repos.pop(session_id, None) - _session_user_ids.pop(session_id, None) - _session_interventions.pop(session_id, None) - _session_error_flags.pop(session_id, None) - _session_article_types.pop(session_id, None) - _session_pending_pause.pop(session_id, None) - - -def _event_type(evt: dict) -> str: - t = evt.get("type", "progress") - # 过滤掉已在上方单独推送的事件类型，避免重复 - if t in ("materials_ready", "outline_ready", "draft_ready", "review_ready", "await_user", - "draft_chunk", "outline_chunk"): - return "step_done" - return t - - -async def _load_graph_snapshot(session_id: str): - """读取 AI 写作 Graph 在该 session 上的状态快照（失败/无状态返回 None）。 - - Graph 的 checkpointer 现在按 config.yaml 持久化到磁盘（默认 SQLite）， - 所以容器重启后只要数据卷还在，仍然能恢复出来。 - """ - try: - from deerflow.agents.ai_writing.graph import get_ai_writing_graph - - 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 - if not snapshot or not getattr(snapshot, "values", None): - return None - return snapshot - - -def _extract_interrupt(snapshot) -> dict | None: - """从快照里取出当前挂起的 interrupt 负载（即 await_user 的数据）。""" - for task in getattr(snapshot, "tasks", None) or (): - for intr in getattr(task, "interrupts", None) or (): - value = getattr(intr, "value", None) - if isinstance(value, dict): - return value - return None - - -async def _recover_session(session_id: str, request: Request | None = None) -> str: - """内存会话注册表丢失时，尝试从 checkpointer 重建会话。 - - 返回: - - "interrupted" —— 图正停在某个干预点，已重建队列并预置 await_user 事件，可继续 resume - - "finished" —— 图已完成，已重建队列并预置 done 事件（+草稿） - - "unrecoverable" —— 无可恢复状态（进程/容器重启过，或 session 从未存在） - - 幂等：内存里已有队列时直接当作 "interrupted" 返回。整段在 `_recovery_lock` - 下执行，避免并发的 /stream 与 /resume 各自重建出不同的队列对象。 - """ - async with _recovery_lock: - if session_id in _event_queues: - return "interrupted" - - snapshot = await _load_graph_snapshot(session_id) - if snapshot is None: - return "unrecoverable" - - queue: asyncio.Queue = asyncio.Queue() - _event_queues[session_id] = queue - _session_interventions.setdefault(session_id, []) - _session_model_names.setdefault(session_id, "") - _session_article_types.setdefault(session_id, []) - - # 尽量恢复 DB 仓库引用，使后续 resume 仍能持久化草稿/审核结果 - if request is not None and session_id not in _session_repos: - try: - from app.gateway.deps import get_ai_writing_session_store, get_current_user - - repo = get_ai_writing_session_store(request) - if repo is not None: - _session_repos[session_id] = repo - _session_user_ids[session_id] = await get_current_user(request) - except Exception: - logger.exception("ai_writing: failed to restore repo during recovery: session=%s", session_id) - - interrupt_val = _extract_interrupt(snapshot) - if interrupt_val is not None: - event = ("await_user", {**interrupt_val, "session_id": session_id}) - await queue.put(event) - _session_pending_pause[session_id] = event - logger.info("ai_writing: recovered interrupted session from checkpointer: session=%s", session_id) - return "interrupted" - - values = snapshot.values or {} - if not getattr(snapshot, "next", None) or values.get("status") == "done": - draft = values.get("current_draft") or {} - if draft.get("full_markdown"): - await queue.put(("draft_ready", { - "session_id": session_id, - "title": draft.get("title", ""), - "markdown": draft.get("full_markdown", ""), - "revision_count": draft.get("revision_count", 0), - })) - await queue.put(("done", {"session_id": session_id, "message": "写作已完成"})) - await queue.put(None) # 哨兵：让 SSE 生成器读完即关闭 - logger.info("ai_writing: recovered finished session from checkpointer: session=%s", session_id) - return "finished" - - # 既不在干预点、也未结束：上次运行多半被中断在某个节点，无法续推 - await queue.put(("error", { - "session_id": session_id, - "message": "写作流程已中断，请重新发起写作。", - })) - await queue.put(None) - logger.warning("ai_writing: session checkpoint stuck mid-flow, not resumable: session=%s", session_id) - return "finished" - - -async def _sse_generator(session_id: str) -> AsyncGenerator[str, None]: - """从队列消费事件，推送为 SSE 帧。""" - # 等待队列创建（最多 5 秒） - for _ in range(50): - if session_id in _event_queues: - break - await asyncio.sleep(0.1) - else: - # 内存里没有这个会话队列。不要直接报「Session not found」—— - # 先尝试从 checkpointer 恢复（图可能仍停在某个干预点 / 已完成）。 - recovery = await _recover_session(session_id) - if recovery == "unrecoverable": - yield _sse("connected", {"session_id": session_id}) - yield _sse("error", { - "session_id": session_id, - "message": "会话状态已丢失（服务可能已重启），请重新发起写作。", - }) - return - # 恢复成功：_event_queues[session_id] 已重建并预置事件，继续走正常消费逻辑。 - - queue = _event_queues.get(session_id) - if queue is None: - # 极端兜底：恢复后又被并发清理 - yield _sse("connected", {"session_id": session_id}) - yield _sse("error", {"session_id": session_id, "message": "会话状态已丢失，请重新发起写作。"}) - return - yield _sse("connected", {"session_id": session_id}) - - # Re-inject the saved await_user event if the queue is empty and a pending pause exists. - # This covers the case where the old SSE generator dequeued the await_user event and - # yielded it to a disconnecting client — the event is then lost from the consume-once queue. - if queue.empty() and session_id in _session_pending_pause: - await queue.put(_session_pending_pause[session_id]) - - while True: - item = await queue.get() - if item is None: - break - event_type, data = item - yield _sse(event_type, data) - await asyncio.sleep(0) diff --git a/offline-backend-20260512/backend/app/gateway/services.py b/offline-backend-20260512/backend/app/gateway/services.py index 4159d635..b5a9b452 100644 --- a/offline-backend-20260512/backend/app/gateway/services.py +++ b/offline-backend-20260512/backend/app/gateway/services.py @@ -100,7 +100,7 @@ _DEFAULT_ASSISTANT_ID = "lead_agent" # Built-in LangGraph graph IDs registered in langgraph.json. # These are NOT stored in the agents DB and must bypass the custom-agent access check. -_BUILTIN_GRAPH_IDS: frozenset[str] = frozenset({"lead_agent", "canvas_agent"}) +_BUILTIN_GRAPH_IDS: frozenset[str] = frozenset({"lead_agent", "canvas_agent", "ai_writing"}) def _prepend_prompt_prefix(graph_input: dict[str, Any], prefix: str) -> dict[str, Any]: @@ -186,15 +186,20 @@ def merge_run_context_overrides(config: dict[str, Any], context: Mapping[str, An def resolve_agent_factory(assistant_id: str | None): """Resolve the agent factory callable from config. - Built-in graphs (``lead_agent``, ``canvas_agent``) are dispatched directly - to their own factory. Custom agents share the ``lead_agent`` factory with - ``agent_id`` injected into configurable/context by :func:`build_run_config`. + Built-in graphs (``lead_agent``, ``canvas_agent``, ``ai_writing``) are dispatched + directly to their own factory. Custom agents share the ``lead_agent`` factory + with ``agent_id`` injected into configurable/context by :func:`build_run_config`. """ if assistant_id == "canvas_agent": from deerflow.agents.canvas_agent.agent import make_canvas_agent return make_canvas_agent + if assistant_id == "ai_writing": + from deerflow.agents.ai_writing.graph import make_ai_writing_graph + + return make_ai_writing_graph + from deerflow.agents.lead_agent.agent import make_lead_agent return make_lead_agent @@ -400,6 +405,11 @@ async def start_run( ) else: await run_ctx.thread_store.update_status(thread_id, "running") + # SDK 的 ``client.threads.create()`` 不传 assistant_id —— 先建出来空的 thread， + # 真正提交 run 时才知道用哪个 assistant。这里回填一次，避免历史列表 / 按 + # assistant 筛选时拿不到关联。 + if not existing.get("assistant_id") and body.assistant_id: + await run_ctx.thread_store.update_assistant_id(thread_id, body.assistant_id) except Exception: logger.warning("Failed to upsert thread_meta for %s (non-fatal)", sanitize_log_param(thread_id)) @@ -441,6 +451,9 @@ async def start_run( stream_subgraphs=body.stream_subgraphs, interrupt_before=body.interrupt_before, interrupt_after=body.interrupt_after, + # 透传 SDK ``Command``：resume / update / goto。Pregel 会据此从最近 + # checkpoint 处恢复，而不是把 graph_input 当新一轮输入跑。 + command=getattr(body, "command", None), ) ) record.task = task diff --git a/offline-backend-20260512/backend/langgraph.json b/offline-backend-20260512/backend/langgraph.json index 00781188..08ea7a34 100644 --- a/offline-backend-20260512/backend/langgraph.json +++ b/offline-backend-20260512/backend/langgraph.json @@ -7,7 +7,8 @@ "env": ".env", "graphs": { "lead_agent": "deerflow.agents:make_lead_agent", - "canvas_agent": "deerflow.agents.canvas_agent:make_canvas_agent" + "canvas_agent": "deerflow.agents.canvas_agent:make_canvas_agent", + "ai_writing": "deerflow.agents.ai_writing.graph:make_ai_writing_graph" }, "auth": { "path": "./app/gateway/langgraph_auth.py:auth" 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 ae2fc5ac..fc822ee1 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,15 +1,3 @@ -from .graph import ( - AI_WRITING_GRAPH, - build_ai_writing_graph, - get_ai_writing_checkpointer, - get_ai_writing_graph, - shutdown_ai_writing_graph, -) +from .graph import make_ai_writing_graph -__all__ = [ - "AI_WRITING_GRAPH", - "build_ai_writing_graph", - "get_ai_writing_graph", - "get_ai_writing_checkpointer", - "shutdown_ai_writing_graph", -] +__all__ = ["make_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 a4c593cc..6cb08c14 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 @@ -3,6 +3,7 @@ from __future__ import annotations import logging +from langchain_core.runnables import RunnableConfig from langgraph.graph import END, StateGraph from langgraph.types import interrupt @@ -15,6 +16,30 @@ from .nodes.editor import editor_node logger = logging.getLogger(__name__) + +async def _mark_session_done(config: RunnableConfig | None) -> None: + """把业务表行 status 写成 ``done``。 + + pause_draft_confirm action=finalize / pause_review_confirm force_finalize 时调， + 让 LangGraph Server 路径下新会话有正确终态（老路径靠 ``_stream_graph_segment`` + finally 块写终态，新路径完全不走它）。失败不影响 graph 流程。 + """ + if not config: + return + thread_id = (config.get("configurable") or {}).get("thread_id") + if not thread_id: + return + from deerflow.persistence.ai_writing_sessions import get_default_store + from deerflow.runtime.user_context import AUTO + + repo = get_default_store() + if repo is None: + return + try: + await repo.update(thread_id, user_id=AUTO, status="done") + except Exception: + logger.exception("ai_writing: mark_session_done failed (non-fatal) thread_id=%s", thread_id) + # ── 干预节点 ────────────────────────────────────────────────────────────────── async def pause_material_confirm(state: AIWritingState) -> dict: @@ -127,7 +152,7 @@ async def pause_outline_confirm(state: AIWritingState) -> dict: return updates -async def pause_draft_confirm(state: AIWritingState) -> dict: +async def pause_draft_confirm(state: AIWritingState, config: RunnableConfig) -> dict: """PAUSE-3：草稿确认""" draft = state["current_draft"] user_input: dict = interrupt({ @@ -141,6 +166,7 @@ async def pause_draft_confirm(state: AIWritingState) -> dict: action = user_input.get("action", "to_editor") if action == "finalize": + await _mark_session_done(config) return { "last_user_intervention": user_input, "status": "done", @@ -223,7 +249,7 @@ async def pause_section_help(state: AIWritingState) -> dict: } -async def pause_review_confirm(state: AIWritingState) -> dict: +async def pause_review_confirm(state: AIWritingState, config: RunnableConfig) -> dict: """PAUSE-4：编辑打回确认""" user_input: dict = interrupt({ "pause_point": "review_confirm", @@ -234,6 +260,7 @@ async def pause_review_confirm(state: AIWritingState) -> dict: }) if user_input.get("overrideVerdict") or user_input.get("action") == "force_finalize": + await _mark_session_done(config) return { "last_user_intervention": user_input, "status": "done", @@ -299,25 +326,26 @@ def after_review_pause(state: AIWritingState) -> str: return "end" if state["status"] == "done" else "writer_draft" -# ── 组装 Graph ──────────────────────────────────────────────────────────────── -def build_ai_writing_graph(checkpointer=None): - """构建并 compile AI 写作 graph。 +# ── LangGraph Server 入口 ──────────────────────────────────────────────────── +# +# compile 出裸 graph（不带 checkpointer），由 LangGraph Server 注入主 checkpointer +# （langgraph.json 里 ``make_checkpointer`` 配的那个）。AI 写作的 thread 状态与 +# 主聊天共享同一份 saver —— LangGraph schema 用 ``(thread_id, checkpoint_ns)`` 做 +# 主键，UUID 不会撞。 +def make_ai_writing_graph(config=None): + """LangGraph Server graph factory：不接收 checkpointer，由 Server 注入。 Args: - checkpointer: 可选的 LangGraph checkpointer。 - - ``None`` → 退回 ``MemorySaver`` —— 仅供测试 / 开发使用。 - 进程或容器重启后会话状态全部丢失，前端拿着旧 session_id 调 - ``/resume`` 会得到 404。 - - 其他 → 传入的 (Async)BaseCheckpointSaver 实例，用于持久化。 + config: LangGraph Server 调用工厂时注入的 RunnableConfig（暂未消费）。 """ g = StateGraph(AIWritingState) - g.add_node("intent_parser", intent_parser_node) - g.add_node("researcher", researcher_node) - g.add_node("pause_material", pause_material_confirm) - g.add_node("writer_outline", writer_outline_node) - g.add_node("pause_outline", pause_outline_confirm) + g.add_node("intent_parser", intent_parser_node) + g.add_node("researcher", researcher_node) + g.add_node("pause_material", pause_material_confirm) + g.add_node("writer_outline", writer_outline_node) + g.add_node("pause_outline", pause_outline_confirm) g.add_node("writer_draft", writer_draft_node) g.add_node("pause_section_help", pause_section_help) g.add_node("pause_draft", pause_draft_confirm) @@ -325,214 +353,23 @@ def build_ai_writing_graph(checkpointer=None): g.add_node("pause_review", pause_review_confirm) g.set_entry_point("intent_parser") - g.add_edge("intent_parser", "researcher") + g.add_edge("intent_parser", "researcher") g.add_conditional_edges("researcher", after_researcher, - {"pause_material": "pause_material", "writer_draft": "writer_draft"}) - + {"pause_material": "pause_material", "writer_draft": "writer_draft"}) g.add_conditional_edges("pause_material", after_material_pause, - {"researcher": "researcher", "writer_outline": "writer_outline"}) - + {"researcher": "researcher", "writer_outline": "writer_outline"}) g.add_edge("writer_outline", "pause_outline") g.add_conditional_edges("pause_outline", after_outline_pause, - {"writer_outline": "writer_outline", "writer_draft": "writer_draft"}) - + {"writer_outline": "writer_outline", "writer_draft": "writer_draft"}) g.add_conditional_edges("writer_draft", after_writer_draft, - {"pause_section_help": "pause_section_help", "pause_draft": "pause_draft"}) - + {"pause_section_help": "pause_section_help", "pause_draft": "pause_draft"}) g.add_conditional_edges("pause_section_help", after_section_help_pause, - {"researcher": "researcher", "writer_draft": "writer_draft"}) - + {"researcher": "researcher", "writer_draft": "writer_draft"}) g.add_conditional_edges("pause_draft", after_draft_pause, - {"end": END, "writer_draft": "writer_draft", "editor": "editor"}) - + {"end": END, "writer_draft": "writer_draft", "editor": "editor"}) g.add_conditional_edges("editor", after_review, - {"pause_draft": "pause_draft", "pause_review": "pause_review"}) - + {"pause_draft": "pause_draft", "pause_review": "pause_review"}) g.add_conditional_edges("pause_review", after_review_pause, - {"end": END, "writer_draft": "writer_draft"}) + {"end": END, "writer_draft": "writer_draft"}) - 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) - - -# ── 持久化 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) + return g.compile() diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/nodes/_utils.py b/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/nodes/_utils.py index 9d2085d7..25236c99 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/nodes/_utils.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/nodes/_utils.py @@ -15,6 +15,29 @@ from langchain_core.runnables import RunnableConfig logger = logging.getLogger(__name__) +def safe_get_stream_writer(): + """返回一个 stream writer callable；在没有 LangGraph runtime 上下文时退化成 no-op。 + + AI 写作节点用它把流式 chunk 推到 LangGraph 原生 ``custom`` 通道。 + 场景： + + - LangGraph Server 路径（``client.runs.stream`` 订阅了 ``stream_mode=["custom", ...]``） + → 真实 writer，事件透传给 SDK 消费方。 + - 老 ``/start /stream /resume`` 路径（``graph.astream(stream_mode="updates")``） + → 真实 writer 仍可调用，只是没人订阅 custom 通道，事件被静默丢弃。 + 这是阶段 2「硬切到 get_stream_writer」的预期行为：老前端 EventSource 不再收到 + token-level chunk，但整体进度卡（draft_ready / review_ready 这些 state 级事件） + 仍正常 —— 阶段 3 前端切到 useStream 后立刻补齐。 + - 单测里直接调节点函数（无 LangGraph runtime） → 返回 no-op，调用方零分支。 + """ + try: + from langgraph.config import get_stream_writer # 延迟 import，便于单测打补丁 + + return get_stream_writer() + except Exception: + return lambda _payload: None + + def get_model(config: RunnableConfig, model_name: str | None = None): """创建对话模型。 diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/nodes/editor.py b/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/nodes/editor.py index 6aa7019b..781c26ff 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/nodes/editor.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/nodes/editor.py @@ -1,8 +1,8 @@ """编辑节点：三项独立 LLM 审查（事实/逻辑/语言），逐 token 流式推送以支持打字机效果。""" from __future__ import annotations -import asyncio import logging +from typing import Callable from langchain_core.messages import HumanMessage, SystemMessage from langchain_core.runnables import RunnableConfig @@ -11,7 +11,7 @@ from ..state import AIWritingState, ReviewResult, ReviewIssue from ..prompts.editor_prompts import ( EDITOR_SYSTEM, FACT_REVIEW_USER, LOGIC_REVIEW_USER, LANGUAGE_REVIEW_USER, ) -from ._utils import get_model, parse_json, _content_to_text, ThinkTagParser +from ._utils import get_model, parse_json, _content_to_text, ThinkTagParser, safe_get_stream_writer logger = logging.getLogger(__name__) @@ -40,9 +40,9 @@ def _parse_issues(raw_issues: list, category: str) -> list[ReviewIssue]: async def _stream_one_review( model, user_prompt: str, category: str, - sse_queue: asyncio.Queue | None, session_id: str, + writer: Callable[[dict], None], ) -> str: - """流式执行单项审查，逐 chunk 推送 review_chunk 事件，返回累积的正文文本（用于 JSON 解析）。""" + """流式执行单项审查，逐 chunk 经 LangGraph custom 通道推送 review_chunk 事件，返回累积的正文文本（用于 JSON 解析）。""" full_text = "" # 仅累积正文（非思考），供后续 JSON 解析 parser = ThinkTagParser() async for chunk in model.astream([ @@ -55,16 +55,42 @@ async def _stream_one_review( for is_thinking, text in parser.feed(delta): if not is_thinking: full_text += text - if sse_queue is not None: - await sse_queue.put(("review_chunk", { - "session_id": session_id, - "category": category, - "text": text, - "is_thinking": is_thinking, - })) + writer({ + "type": "review_chunk", + "category": category, + "text": text, + "is_thinking": is_thinking, + }) return full_text +async def _save_review_to_repo(config: RunnableConfig, review: ReviewResult) -> None: + """把 review_result 异步落到业务表（best-effort，写失败不影响流程）。 + + ``user_id`` 走 ``resolve_user_id(AUTO)`` 读 contextvar（与 intent_parser 建行、 + writer_draft 写草稿同套规则）。 + """ + cfg = config.get("configurable", {}) or {} + thread_id = cfg.get("thread_id") + if not thread_id: + return + from deerflow.persistence.ai_writing_sessions import get_default_store + from deerflow.runtime.user_context import AUTO + + repo = get_default_store() + if repo is None: + return + try: + await repo.update( + thread_id, + user_id=AUTO, + status="review_ready", + review_result=dict(review), + ) + except Exception: + logger.exception("editor: save_review failed (non-fatal) thread_id=%s", thread_id) + + async def editor_node(state: AIWritingState, config: RunnableConfig) -> dict: draft = state.get("current_draft") if not draft: @@ -78,8 +104,7 @@ async def editor_node(state: AIWritingState, config: RunnableConfig) -> dict: mats = _materials_summary(pkg) cfg = config.get("configurable", {}) - sse_queue: asyncio.Queue | None = cfg.get("sse_queue") - session_id: str = cfg.get("sse_session_id", "") + writer = safe_get_stream_writer() ai_writing_config = cfg.get("ai_writing_config", {}) or {} pass_threshold = int(ai_writing_config.get("pass_threshold", _DEFAULT_PASS_THRESHOLD)) @@ -110,7 +135,7 @@ async def editor_node(state: AIWritingState, config: RunnableConfig) -> dict: for category, user_prompt in REVIEW_TASKS: fallback = {"score": 80, "issues": []} try: - full_text = await _stream_one_review(model, user_prompt, category, sse_queue, session_id) + full_text = await _stream_one_review(model, user_prompt, category, writer) res = parse_json(full_text, fallback) except Exception as e: logger.warning("editor review [%s] failed: %s", category, e) @@ -120,16 +145,13 @@ async def editor_node(state: AIWritingState, config: RunnableConfig) -> dict: results[category] = res # 这一项审查完成 —— 推送结构化结果（前端据此把流式文本替换为评分/意见） - if sse_queue is not None: - issues = _parse_issues(res.get("issues", []), category) - await sse_queue.put(("review_item_done", { - "session_id": session_id, - "category": category, - "score": int(res.get("score", 80)), - "issues": [dict(iss) for iss in issues], - })) - else: - logger.debug("editor [%s] done (no sse_queue)", category) + issues = _parse_issues(res.get("issues", []), category) + writer({ + "type": "review_item_done", + "category": category, + "score": int(res.get("score", 80)), + "issues": [dict(iss) for iss in issues], + }) # 汇总 fact_score = int(results["fact"].get("score", 80)) @@ -156,6 +178,9 @@ async def editor_node(state: AIWritingState, config: RunnableConfig) -> dict: revision_notes=all_issues, ) + # 阶段 2：节点内并存 DB 写入（路由层同样也写一遍，阶段 7 删冗余时移除路由那份） + await _save_review_to_repo(config, review) + # 分项审查出现模型异常时，以非致命 notice 提示用户，审核结论仍照常产出 notices: list[dict] = [] if review_errors: diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/nodes/intent_parser.py b/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/nodes/intent_parser.py index 92825d68..a8345734 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/nodes/intent_parser.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/nodes/intent_parser.py @@ -1,6 +1,7 @@ """意图解析节点：将用户自然语言意图结构化为写作配置。""" from __future__ import annotations +import logging import re from langchain_core.runnables import RunnableConfig @@ -8,6 +9,49 @@ from langchain_core.runnables import RunnableConfig from ..state import AIWritingState from ._utils import llm_json +logger = logging.getLogger(__name__) + + +def _truncate_title(text: str, max_len: int = 50) -> str: + """与老 /start 路由里 _truncate_title 行为一致：UTF 字符截断 + 省略号。""" + text = (text or "").strip() + if len(text) <= max_len: + return text + return text[:max_len].rstrip() + "…" + + +async def _upsert_business_row(configurable: dict, state: AIWritingState) -> None: + """在 intent_parser 入口幂等创建 ai_writing_sessions 业务表行。 + + LangGraph Server 路径下没有 /start 路由建行，必须节点自建，否则列表页查不到、 + transcript / draft / review 后续写入都会因 update 找不到行而静默丢数据。 + resume 路径下 thread_id 已经有行，upsert_session 只刷新 updated_at。 + + user_id 走 ``resolve_user_id(AUTO)``：FastAPI 请求线程把它写到 contextvar， + asyncio.create_task 继承父任务的 context，所以这里能拿到当前用户； + contextvar 没设时（测试 / 离线场景）抛 RuntimeError，try/except 兜底跳过。 + """ + from deerflow.persistence.ai_writing_sessions import get_default_store + from deerflow.runtime.user_context import AUTO + + thread_id = configurable.get("thread_id") + if not thread_id: + return + repo = get_default_store() + if repo is None: + return + user_intent = (state.get("user_intent") or "").strip() + try: + await repo.upsert_session( + id=thread_id, + title=_truncate_title(user_intent) or "AI 写作", + user_intent=user_intent, + user_id=AUTO, + status="in_progress", + ) + except Exception: + logger.exception("intent_parser: upsert_session failed (non-fatal) thread_id=%s", thread_id) + # 兜底类型列表，当 config 中未注入 valid_article_types 时使用 _DEFAULT_ARTICLE_TYPES: list[dict] = [ {"key": "news", "label": "新闻资讯"}, @@ -79,6 +123,10 @@ async def intent_parser_node(state: AIWritingState, config: RunnableConfig) -> d valid_keys: list[str] = [t["key"] for t in valid_article_types] default_key: str = valid_keys[0] if valid_keys else "analysis" + # 业务表 upsert（LangGraph Server 路径下没有 /start 路由建行，节点入口自建； + # resume 时幂等不重复建。失败不阻断写作流程）。 + await _upsert_business_row(configurable, state) + type_options = _build_type_options(valid_article_types) system = _SYSTEM_TEMPLATE.format(type_options=type_options) user = _USER_TEMPLATE.format( diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/nodes/researcher.py b/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/nodes/researcher.py index ab4691dd..79c6c9c9 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/nodes/researcher.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/nodes/researcher.py @@ -15,7 +15,7 @@ from ..prompts.researcher_prompts import ( EXTRACT_KEYWORDS_SYSTEM, EXTRACT_KEYWORDS_USER, RANK_MATERIALS_SYSTEM, RANK_MATERIALS_USER, ) -from ._utils import llm_json +from ._utils import llm_json, safe_get_stream_writer logger = logging.getLogger(__name__) @@ -103,10 +103,9 @@ async def researcher_node(state: AIWritingState, config: RunnableConfig) -> dict # 用户指定的检索关键词数量（1~10），决定生成多少个检索词、检索多少个方向 keyword_count = _resolve_keyword_count(state.get("keyword_count")) - # SSE 实时进度推送（与 outline/draft 节点共用 sse_queue 机制） + # 实时进度推送：阶段 2 起统一走 LangGraph 原生 custom 通道。 cfg = config.get("configurable", {}) - sse_queue = cfg.get("sse_queue") - sse_session_id: str = cfg.get("sse_session_id", "") + writer = safe_get_stream_writer() # 素材收集专家专用模型：检索词生成 / 素材排序 可在 config.yaml 的 # ai_writing 段单独配置（keyword_model / rank_model），留空则回退写作主模型。 @@ -116,16 +115,11 @@ async def researcher_node(state: AIWritingState, config: RunnableConfig) -> dict # 素材总量上限（config.yaml: ai_writing.max_materials），按检索词数均分到每个词 max_materials = _resolve_max_materials(ai_writing_config.get("max_materials")) - async def _emit(message: str) -> None: - if sse_queue is None: - return - await sse_queue.put(("research_progress", { - "session_id": sse_session_id, - "message": message, - })) + def _emit(message: str) -> None: + writer({"type": "research_progress", "message": message}) # Step 1: 提取关键词 - await _emit("正在生成检索词…") + _emit("正在生成检索词…") kw_result = await llm_json( EXTRACT_KEYWORDS_SYSTEM, EXTRACT_KEYWORDS_USER.format( @@ -154,7 +148,7 @@ async def researcher_node(state: AIWritingState, config: RunnableConfig) -> dict total = len(search_keywords) # 每个检索词分到的检索条数 = 总量上限 / 检索词数（向上取整） per_keyword = _per_keyword_limit(max_materials, total) - await _emit(f"检索词已生成，开始检索 {total} 个方向…") + _emit(f"检索词已生成，开始检索 {total} 个方向…") # 并发搜索，但每完成一个就推送进度（计数在单线程 asyncio 下无需加锁） completed = 0 @@ -166,9 +160,9 @@ async def researcher_node(state: AIWritingState, config: RunnableConfig) -> dict finally: completed += 1 if completed < total: - await _emit(f"已完成 {completed}/{total} 次检索，正在继续检索相关内容…") + _emit(f"已完成 {completed}/{total} 次检索，正在继续检索相关内容…") else: - await _emit(f"已完成全部 {total} 次检索") + _emit(f"已完成全部 {total} 次检索") search_tasks = [_search_one(kw) for kw in search_keywords] all_results_nested: list[list[SearchResult]] = await asyncio.gather(*search_tasks, return_exceptions=True) @@ -208,7 +202,7 @@ async def researcher_node(state: AIWritingState, config: RunnableConfig) -> dict search_errors.extend(getattr(provider, "errors", [])) # Step 3: 相关度排序（如素材较多，让 LLM 排序） - await _emit("检索完成，正在筛选排序素材…") + _emit("检索完成，正在筛选排序素材…") rank_error = "" if len(raw_materials) > max_materials: materials_list = "\n".join( diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/nodes/writer_draft.py b/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/nodes/writer_draft.py index d3fdcb8c..c172541d 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/nodes/writer_draft.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/nodes/writer_draft.py @@ -1,7 +1,6 @@ """作家节点 Phase 2：按大纲逐段写正文，支持流式 chunk 输出。""" from __future__ import annotations -import asyncio import logging import re @@ -13,7 +12,7 @@ from ..prompts.writer_prompts import ( DRAFT_SECTION_SYSTEM, DRAFT_SECTION_USER, DRAFT_EXPAND_USER, WRITER_MODE_LOOSE, WRITER_MODE_STRICT, SECTION_BLOCKED_MARKER, ) -from ._utils import get_model, _content_to_text, count_words, ThinkTagParser +from ._utils import get_model, _content_to_text, count_words, ThinkTagParser, safe_get_stream_writer logger = logging.getLogger(__name__) @@ -64,6 +63,35 @@ def _get_section_materials(material_ids: list[str], pkg: dict) -> str: return "\n".join(lines) if lines else "（无相关素材）" +async def _save_draft_to_repo(config: RunnableConfig, draft: DraftArticle) -> None: + """把 current_draft 异步落到业务表（best-effort，写失败不影响流程）。 + + ``user_id`` 走 ``resolve_user_id(AUTO)`` 读 contextvar（FastAPI 请求线程设置， + asyncio.create_task 继承）。intent_parser 在阶段 4 已用同样规则建行；这里 update + 时也用 AUTO 才能命中 owner filter。 + """ + cfg = config.get("configurable", {}) or {} + thread_id = cfg.get("thread_id") + if not thread_id: + return + from deerflow.persistence.ai_writing_sessions import get_default_store + from deerflow.runtime.user_context import AUTO + + repo = get_default_store() + if repo is None: + return + try: + await repo.update( + thread_id, + user_id=AUTO, + status="draft_ready", + draft_title=draft.get("title") or "", + draft_markdown=draft.get("full_markdown") or "", + ) + except Exception: + logger.exception("writer_draft: save_draft failed (non-fatal) thread_id=%s", thread_id) + + async def writer_draft_node(state: AIWritingState, config: RunnableConfig) -> dict: outline = state.get("current_outline") if not outline: @@ -80,10 +108,10 @@ async def writer_draft_node(state: AIWritingState, config: RunnableConfig) -> di writing_mode = state.get("writing_mode", "loose") loose_titles: set[str] = set(intervention.get("_loose_sections") or []) - # SSE 流式推送所需 + # 流式推送：阶段 2 起统一走 LangGraph 原生 custom 通道（get_stream_writer）。 + # cfg 仍保留以便阶段 4 取 user_id / thread_id 用。 cfg = config.get("configurable", {}) - sse_queue: asyncio.Queue | None = cfg.get("sse_queue") - session_id: str = cfg.get("sse_session_id", "") + writer = safe_get_stream_writer() # 修改意见（来自用户或编辑） revision_notes = "" @@ -174,25 +202,23 @@ async def writer_draft_node(state: AIWritingState, config: RunnableConfig) -> di for is_thinking, text in parser.feed(delta): if not is_thinking: section_buf += text - if sse_queue is not None: - await sse_queue.put(("draft_chunk", { - "session_id": session_id, - "section_index": i, - "section_title": sec_title, - "text": text, - "is_thinking": is_thinking, - "is_section_done": False, - })) + writer({ + "type": "draft_chunk", + "section_index": i, + "section_title": sec_title, + "text": text, + "is_thinking": is_thinking, + "is_section_done": False, + }) # 章节完成 - if sse_queue is not None: - await sse_queue.put(("draft_chunk", { - "session_id": session_id, - "section_index": i, - "section_title": sec_title, - "text": "", - "is_section_done": True, - })) + writer({ + "type": "draft_chunk", + "section_index": i, + "section_title": sec_title, + "text": "", + "is_section_done": True, + }) cleaned_buf = section_buf.strip() blocked_reason = "" @@ -289,6 +315,11 @@ async def writer_draft_node(state: AIWritingState, config: RunnableConfig) -> di revision_count=revision_count, ) + # 阶段 2：节点内并存 DB 写入（路由层 _stream_graph_segment 同样也会写一遍， + # 阶段 7 删冗余时移除路由层那份）。new path (LangGraph Server) 下路由层 + # 不再介入，节点写就是唯一来源。 + await _save_draft_to_repo(config, draft) + # 分节写作出现模型异常时，以非致命 notice 提示用户，草稿仍照常产出 notices: list[dict] = [] if section_errors: diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/nodes/writer_outline.py b/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/nodes/writer_outline.py index 15f149fc..82108575 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/nodes/writer_outline.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/nodes/writer_outline.py @@ -1,7 +1,6 @@ """作家节点 Phase 1：生成文章大纲，支持流式 chunk 输出。""" from __future__ import annotations -import asyncio import json import logging @@ -12,7 +11,7 @@ from ..state import AIWritingState, Outline, OutlineSection from ..prompts.writer_prompts import ( OUTLINE_SYSTEM, OUTLINE_USER, OUTLINE_REVISE_SYSTEM, OUTLINE_REVISE_USER, ) -from ._utils import get_model, parse_json, _content_to_text, ThinkTagParser +from ._utils import get_model, parse_json, _content_to_text, ThinkTagParser, safe_get_stream_writer logger = logging.getLogger(__name__) @@ -79,10 +78,8 @@ async def writer_outline_node(state: AIWritingState, config: RunnableConfig) -> feedback_section=feedback_section, ) - # SSE 流式推送 - cfg = config.get("configurable", {}) - sse_queue: asyncio.Queue | None = cfg.get("sse_queue") - session_id: str = cfg.get("sse_session_id", "") + # 流式推送：阶段 2 起统一走 LangGraph 原生 custom 通道（get_stream_writer）。 + writer = safe_get_stream_writer() full_text = "" # 只累积正文（非思考）用于 JSON 解析 parser = ThinkTagParser() @@ -101,12 +98,11 @@ async def writer_outline_node(state: AIWritingState, config: RunnableConfig) -> for is_thinking, text in parser.feed(delta): if not is_thinking: full_text += text - if sse_queue is not None: - await sse_queue.put(("outline_chunk", { - "session_id": session_id, - "text": text, - "is_thinking": is_thinking, - })) + writer({ + "type": "outline_chunk", + "text": text, + "is_thinking": is_thinking, + }) except Exception as e: logger.error("outline streaming failed: %s", e, exc_info=True) return { diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/ai_writing_sessions/__init__.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/ai_writing_sessions/__init__.py index 7b1e573f..53d0a5ec 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/ai_writing_sessions/__init__.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/ai_writing_sessions/__init__.py @@ -8,4 +8,39 @@ def make_ai_writing_session_store(session_factory) -> AIWritingSessionRepository return AIWritingSessionRepository(session_factory) -__all__ = ["AIWritingSessionRow", "AIWritingSessionRepository", "make_ai_writing_session_store"] +# ── 进程级默认 repo 单例 ────────────────────────────────────────────────────── +# +# AI 写作 graph 注册成 LangGraph Server assistant 后，节点跑在 harness 进程里， +# 拿不到 FastAPI 的 ``request.app.state``；harness → app 又是禁止的 import 方向。 +# 这里提供一个 app 启动时注册、harness 节点运行时读取的「跨边界单例」通道。 +# +# 写入：app/gateway/deps.py 在创建完 session_factory 之后调 +# ``register_default_store(app.state.ai_writing_session_store)`` +# 读取：节点里 ``get_default_store()``，None 表示未配置 DB（跳过写入即可）。 +_default_store: AIWritingSessionRepository | None = None + + +def register_default_store(store: AIWritingSessionRepository | None) -> None: + """登记进程级默认 AI 写作 session store。 + + 幂等：可多次调用，覆盖前值。传 ``None`` 用于显式清理（主要给测试）。 + """ + global _default_store + _default_store = store + + +def get_default_store() -> AIWritingSessionRepository | None: + """返回进程级默认 AI 写作 session store；未配置时为 ``None``。 + + 节点调用方应在拿到 ``None`` 时跳过 DB 写入（写入是「锦上添花」非致命）。 + """ + return _default_store + + +__all__ = [ + "AIWritingSessionRow", + "AIWritingSessionRepository", + "make_ai_writing_session_store", + "register_default_store", + "get_default_store", +] diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/ai_writing_sessions/model.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/ai_writing_sessions/model.py index b4891ce7..14377740 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/ai_writing_sessions/model.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/ai_writing_sessions/model.py @@ -4,11 +4,11 @@ from __future__ import annotations from datetime import UTC, datetime -from sqlalchemy import DateTime, String, Text +from sqlalchemy import String, Text from sqlalchemy.orm import Mapped, mapped_column from deerflow.persistence.base import Base -from deerflow.persistence.types import PortableLongText +from deerflow.persistence.types import BeijingDateTime, PortableLongText class AIWritingSessionRow(Base): @@ -26,5 +26,13 @@ class AIWritingSessionRow(Base): # 完整对话时间线（前端组装的 progressEvents + completedInterventions），用于历史回显。 # 用 PortableLongText：MySQL 上映射为 LONGTEXT，避免 TEXT 64KB 上限导致保存 500。 transcript: Mapped[str | None] = mapped_column(PortableLongText, nullable=True) # JSON object - created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=lambda: datetime.now(UTC)) - updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=lambda: datetime.now(UTC)) + # 时间列必须用 BeijingDateTime（项目规范，与 threads_meta 等其它表对齐）—— MySQL + # DATETIME 不存时区，raw ``DateTime(timezone=True)`` 会把 UTC wall clock 写入 + # 列，读出来比真实北京时间晚 8 小时。``BeijingDateTime`` 在 bind 时 UTC→Beijing、 + # 在 result 时 attach +08:00，前后端拿到的就是北京时间且 tz-aware。 + created_at: Mapped[datetime] = mapped_column(BeijingDateTime(), default=lambda: datetime.now(UTC)) + updated_at: Mapped[datetime] = mapped_column( + BeijingDateTime(), + default=lambda: datetime.now(UTC), + onupdate=lambda: datetime.now(UTC), + ) 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 b6c4306e..30b86f5c 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 @@ -63,6 +63,52 @@ class AIWritingSessionRepository: await session.refresh(row) return self._row_to_dict(row) + async def upsert_session( + self, + *, + id: str, + title: str, + user_intent: str | None = None, + user_id: str | None | _AutoSentinel = AUTO, + status: str = "in_progress", + ) -> dict: + """幂等：行不存在则创建，行存在则保留原数据（不覆盖 draft / review 等已有写入）。 + + AI 写作 graph 的 ``intent_parser_node`` 在每次进入时调一次（首启 + resume 时 + 都会跑），所以必须幂等。resume 时若直接 ``create`` 会因主键冲突抛错；用 + ``update`` 又找不到行而 no-op。这个方法把两条路合一：行不存在新建，行存在 + 只刷新 ``updated_at`` 以便列表页排序前置。 + """ + resolved_user_id = resolve_user_id(user_id, method_name="AIWritingSessionRepository.upsert_session") + now = datetime.now(UTC) + async with self._sf() as session: + row = await session.get(AIWritingSessionRow, id) + if row is None: + row = AIWritingSessionRow( + id=id, + user_id=resolved_user_id, + title=title, + user_intent=user_intent, + status=status, + draft_title=None, + draft_markdown=None, + completed_interventions=None, + review_result=None, + transcript=None, + created_at=now, + updated_at=now, + ) + session.add(row) + else: + # 行已存在（resume 路径）：仅刷新 updated_at，保留 user_id / title / draft / review 等。 + # 若 user_id 之前为空（migration 兼容场景），用解析出来的填上。 + if row.user_id is None and resolved_user_id is not None: + row.user_id = resolved_user_id + row.updated_at = now + await session.commit() + await session.refresh(row) + return self._row_to_dict(row) + async def update( self, session_id: str, diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/thread_meta/base.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/thread_meta/base.py index 85c5f2cd..29a89ed7 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/thread_meta/base.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/thread_meta/base.py @@ -56,6 +56,17 @@ class ThreadMetaStore(abc.ABC): async def update_status(self, thread_id: str, status: str, *, user_id: str | None | _AutoSentinel = AUTO) -> None: pass + @abc.abstractmethod + async def update_assistant_id(self, thread_id: str, assistant_id: str, *, user_id: str | None | _AutoSentinel = AUTO) -> None: + """Backfill ``assistant_id`` for a thread row. + + Used by ``start_run`` when the row was pre-created via ``POST /threads`` + without an assistant_id (the SDK's ``client.threads.create()`` doesn't + pass it), and the first run on the thread reveals the assistant. No-op + if the thread does not exist or the owner check fails. + """ + pass + @abc.abstractmethod async def update_metadata(self, thread_id: str, metadata: dict, *, user_id: str | None | _AutoSentinel = AUTO) -> None: """Merge ``metadata`` into the thread's metadata field. diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/thread_meta/memory.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/thread_meta/memory.py index 99afc997..b5d8a876 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/thread_meta/memory.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/thread_meta/memory.py @@ -117,6 +117,14 @@ class MemoryThreadMetaStore(ThreadMetaStore): record["updated_at"] = now_iso() await self._store.aput(THREADS_NS, thread_id, record) + async def update_assistant_id(self, thread_id: str, assistant_id: str, *, user_id: str | None | _AutoSentinel = AUTO) -> None: + record = await self._get_owned_record(thread_id, user_id, "MemoryThreadMetaStore.update_assistant_id") + if record is None: + return + record["assistant_id"] = assistant_id + record["updated_at"] = now_iso() + await self._store.aput(THREADS_NS, thread_id, record) + async def update_metadata(self, thread_id: str, metadata: dict, *, user_id: str | None | _AutoSentinel = AUTO) -> None: record = await self._get_owned_record(thread_id, user_id, "MemoryThreadMetaStore.update_metadata") if record is None: diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/thread_meta/sql.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/thread_meta/sql.py index 08389be7..c9a70f31 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/thread_meta/sql.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/thread_meta/sql.py @@ -174,6 +174,20 @@ class ThreadMetaRepository(ThreadMetaStore): await session.execute(update(ThreadMetaRow).where(ThreadMetaRow.thread_id == thread_id).values(status=status, updated_at=datetime.now(UTC))) await session.commit() + async def update_assistant_id( + self, + thread_id: str, + assistant_id: str, + *, + user_id: str | None | _AutoSentinel = AUTO, + ) -> None: + resolved_user_id = resolve_user_id(user_id, method_name="ThreadMetaRepository.update_assistant_id") + async with self._sf() as session: + if not await self._check_ownership(session, thread_id, resolved_user_id): + return + await session.execute(update(ThreadMetaRow).where(ThreadMetaRow.thread_id == thread_id).values(assistant_id=assistant_id, updated_at=datetime.now(UTC))) + await session.commit() + async def update_metadata( self, thread_id: str, diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/runtime/runs/worker.py b/offline-backend-20260512/backend/packages/harness/deerflow/runtime/runs/worker.py index 0a6b188a..025fd95d 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/runtime/runs/worker.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/runtime/runs/worker.py @@ -130,8 +130,16 @@ async def run_agent( stream_subgraphs: bool = False, interrupt_before: list[str] | Literal["*"] | None = None, interrupt_after: list[str] | Literal["*"] | None = None, + command: dict[str, Any] | None = None, ) -> None: - """Execute an agent in the background, publishing events to *bridge*.""" + """Execute an agent in the background, publishing events to *bridge*. + + ``command`` 来自 LangGraph SDK 的 ``Command`` 协议（``{resume, update, goto}``）。 + 存在时把它翻译成 ``langgraph.types.Command`` 并作为 ``astream`` 的入参 —— + Pregel 检查到 Command 输入会从最近 checkpoint 处恢复（典型场景：interrupt 之后 + 的 ``Command(resume=user_input)``）；不存在时仍按 ``graph_input`` dict 起新 run。 + AI 写作的 4 个干预卡就靠这个把 user_input 传回 ``pause_*_confirm`` 节点。 + """ # Unpack infrastructure dependencies from RunContext. checkpointer = ctx.checkpointer @@ -269,10 +277,23 @@ async def run_agent( logger.info("Run %s: streaming with modes %s (requested: %s)", run_id, lg_modes, requested_modes) # 7. Stream using graph.astream + # 如果客户端传了 Command（典型：interrupt 后续提的 ``Command(resume=...)``）， + # 把它翻译成 langgraph.types.Command 作为 astream 输入；否则按 graph_input dict + # 起新一轮。Pregel 检查到 Command 会从最近 checkpoint 处恢复，而不是重头跑。 + stream_input: Any = graph_input + if command is not None: + from langgraph.types import Command as _LGCommand + + stream_input = _LGCommand( + resume=command.get("resume"), + update=command.get("update"), + goto=command.get("goto"), + ) + if len(lg_modes) == 1 and not stream_subgraphs: # Single mode, no subgraphs: astream yields raw chunks single_mode = lg_modes[0] - async for chunk in agent.astream(graph_input, config=runnable_config, stream_mode=single_mode): + async for chunk in agent.astream(stream_input, config=runnable_config, stream_mode=single_mode): if record.abort_event.is_set(): logger.info("Run %s abort requested — stopping", run_id) break @@ -281,7 +302,7 @@ async def run_agent( else: # Multiple modes or subgraphs: astream yields tuples async for item in agent.astream( - graph_input, + stream_input, config=runnable_config, stream_mode=lg_modes, subgraphs=stream_subgraphs, diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/runtime/serialization.py b/offline-backend-20260512/backend/packages/harness/deerflow/runtime/serialization.py index b3835d42..438d1d94 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/runtime/serialization.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/runtime/serialization.py @@ -10,6 +10,7 @@ Consumers: ``deerflow.runtime.runs.worker`` (SSE publishing) and from __future__ import annotations +import dataclasses from typing import Any @@ -35,6 +36,16 @@ def serialize_lc_object(obj: Any) -> Any: return obj.dict() except Exception: pass + # Dataclass instances —— LangGraph 的 ``Interrupt`` 就是 dataclass 没有 + # model_dump 也没有 dict 方法；之前 fallback 到 ``str(obj)`` 会把 + # ``Interrupt(value=..., id='...')`` 序列化成字符串，下游 SDK + # (``@langchain/langgraph-sdk``) 期待 dict 结构 → 直接丢弃 → AI 写作 + # 的 4 个干预卡永远不弹（参 ai-writing-优化.md §阶段 3 排查记录）。 + if dataclasses.is_dataclass(obj) and not isinstance(obj, type): + try: + return {k: serialize_lc_object(v) for k, v in dataclasses.asdict(obj).items()} + except Exception: + pass # Last resort try: return str(obj) @@ -45,12 +56,21 @@ def serialize_lc_object(obj: Any) -> Any: def serialize_channel_values(channel_values: dict[str, Any]) -> dict[str, Any]: """Serialize channel values, stripping internal LangGraph keys. - Internal keys like ``__pregel_*`` and ``__interrupt__`` are removed - to match what the LangGraph Platform API returns. + ``__pregel_*`` 键始终剥离。``__interrupt__`` 仅在**为空时**剥离 —— 因为 + ``@langchain/langgraph-sdk`` 的 ``useStream`` 只在 ``event === "values"`` + 通道里读 ``data.__interrupt__`` 并通过 ``stream.interrupt`` 暴露给上层。 + AI 写作的 4 个干预卡（material/outline/draft/review）就是靠它触发的； + 继续过滤会让前端永远拿不到 interrupt。 + + 主聊天不消费 ``stream.interrupt``，但若我们让一个空数组溜过去，SDK 会 + fallback 成 ``{when: "breakpoint"}`` 的伪 interrupt —— 所以仅在数组非空 + 时保留，两侧都不破坏。 """ result: dict[str, Any] = {} for key, value in channel_values.items(): - if key.startswith("__pregel_") or key == "__interrupt__": + if key.startswith("__pregel_"): + continue + if key == "__interrupt__" and not value: continue result[key] = serialize_lc_object(value) return result diff --git a/offline-backend-20260512/backend/scripts/probe_ai_writing_assistant.py b/offline-backend-20260512/backend/scripts/probe_ai_writing_assistant.py new file mode 100644 index 00000000..2733a885 --- /dev/null +++ b/offline-backend-20260512/backend/scripts/probe_ai_writing_assistant.py @@ -0,0 +1,179 @@ +"""AI 写作 assistant 冒烟脚本：验证 ai_writing graph 注册到 LangGraph Server 后， +interrupt/Command(resume=...)/state 持久化/custom stream 全链路是否跑通。 + +阶段 0 用名为 ai_writing_probe；阶段 1 起改为正式 assistant 名 ai_writing，本脚本随之切换。 + +前置条件： + 1. langgraph.json 已注册 ai_writing（见 ai-writing-优化.md 阶段 1）。 + 2. 后端已通过 `make dev` 启动，Gateway + LangGraph Server 监听 8001。 + 3. config.yaml 配好至少一个可用 model，researcher 用到的 search_provider 也已配置 + —— 否则 researcher 节点会失败（这是业务依赖，不是迁移问题，仍然算「跑通」）。 + +运行： + cd offline-backend-20260512/backend + PYTHONPATH=. uv run python scripts/probe_ai_writing_assistant.py + +通过条件（与 ai-writing-优化.md §阶段 0 §通过条件 对齐）： + - assistants 列表里能看到 ai_writing_probe + - 跑到 pause_material_confirm 能拿到 interrupt 事件 + - Command(resume={"action": "confirm", ...}) 能成功续跑 + - 重建客户端后 client.threads.get_state(thread_id) 仍能拿到上一阶段 state +""" +from __future__ import annotations + +import asyncio +import os +import sys +from typing import Any + +try: + from langgraph_sdk import get_client +except ImportError: + print("ERROR: langgraph-sdk 未安装。run: uv pip install langgraph-sdk", file=sys.stderr) + sys.exit(1) + + +ASSISTANT_ID = "ai_writing" +DEFAULT_URL = os.environ.get("DEERFLOW_LANGGRAPH_URL", "http://localhost:8001/api/langgraph") + + +def _banner(text: str) -> None: + print(f"\n{'=' * 8} {text} {'=' * 8}") + + +def _summarize_event(event: Any) -> str: + """裁剪日志，避免一屏全是大段 markdown / 素材列表。""" + s = repr(event) + return s if len(s) <= 240 else s[:240] + "…" + + +async def _stream_until_interrupt( + client, thread_id: str, *, input_payload: dict | None = None, command: dict | None = None +) -> dict | None: + """跑一次 runs.stream，捕获第一个 interrupt（含 pause payload）后返回，否则返回 None。 + + 收到 interrupt 时把 payload 一并打出来，便于人工核对 4 个 pause 点的 schema。 + """ + interrupt_payload: dict | None = None + kwargs: dict[str, Any] = { + "stream_mode": ["values", "messages-tuple", "custom", "updates"], + } + if input_payload is not None: + kwargs["input"] = input_payload + if command is not None: + kwargs["command"] = command + + async for chunk in client.runs.stream(thread_id, ASSISTANT_ID, **kwargs): + # langgraph-sdk 把不同 mode 的事件混在同一 stream 里，按 chunk.event 分类。 + event_name = getattr(chunk, "event", None) or ( + chunk.get("event") if isinstance(chunk, dict) else None + ) + data = getattr(chunk, "data", None) or ( + chunk.get("data") if isinstance(chunk, dict) else None + ) + print(f"[stream] event={event_name} data={_summarize_event(data)}") + + # SDK 在遇到 interrupt() 时会发一个特殊事件 —— 不同版本字段名不同，做下兼容。 + if event_name == "updates" and isinstance(data, dict): + for _node, node_update in data.items(): + if isinstance(node_update, dict) and "__interrupt__" in node_update: + interrupt_payload = node_update["__interrupt__"] + if event_name in {"interrupt", "__interrupt__"}: + interrupt_payload = data + break + + return interrupt_payload + + +async def main() -> int: + print(f"LangGraph URL = {DEFAULT_URL}") + client = get_client(url=DEFAULT_URL) + + # ── Step 1: 确认 assistant 已注册 ────────────────────────────────────────── + _banner("Step 1: 列 assistant，确认 probe 已注册") + assistants = await client.assistants.search() + names = [a.get("name") or a.get("graph_id") for a in assistants] + print(f"已注册 assistants: {names}") + if not any(ASSISTANT_ID in (n or "") for n in names): + print(f"FAIL: 未找到 {ASSISTANT_ID}，请检查 langgraph.json 注册是否生效。") + return 1 + + # ── Step 2: 创建 thread 并触发第一次 run ────────────────────────────────── + _banner("Step 2: 创建 thread + 启动写作（应停在 pause_material）") + thread = await client.threads.create() + thread_id = thread["thread_id"] + print(f"thread_id = {thread_id}") + + start_input = { + "user_intent": "写一篇关于 LangGraph 多 worker 部署最佳实践的分析报道，目标读者是后端工程师，约 800 字", + "article_type": "analysis", + "target_audience": "后端工程师", + "word_count_target": 800, + "keyword_count": 4, + "writing_mode": "loose", + "max_revisions": 2, + "status": "researching", + "drafts": [], + "progress_events": [], + } + payload = await _stream_until_interrupt(client, thread_id, input_payload=start_input) + if not payload: + print("FAIL: 未收到 interrupt 事件 —— pause_material 可能没停住，或 researcher 抛错。") + return 2 + print(f"\nPAUSE-1 payload = {_summarize_event(payload)}") + + # ── Step 3: 确认素材，应停在 pause_outline ──────────────────────────────── + _banner("Step 3: Command(resume) 确认素材（应停在 pause_outline）") + payload = await _stream_until_interrupt( + client, thread_id, + command={"resume": {"action": "confirm"}}, # 不指定 approvedMaterialIds → 默认全部通过 + ) + if not payload: + print("FAIL: 未停在 pause_outline。") + return 3 + print(f"\nPAUSE-2 payload = {_summarize_event(payload)}") + + # ── Step 4: 确认大纲，应停在 pause_draft 或 pause_section_help ─────────── + _banner("Step 4: Command(resume) 确认大纲（应停在 pause_draft 或 pause_section_help）") + payload = await _stream_until_interrupt( + client, thread_id, + command={"resume": {"action": "confirm"}}, + ) + if not payload: + print("FAIL: 未停在 pause_draft / pause_section_help。") + return 4 + print(f"\nPAUSE-3 payload = {_summarize_event(payload)}") + + # ── Step 5: 定稿 ───────────────────────────────────────────────────────── + _banner("Step 5: Command(resume) finalize 草稿（应跑到 END）") + payload = await _stream_until_interrupt( + client, thread_id, + command={"resume": {"action": "finalize"}}, + ) + if payload: + print(f"NOTE: 收到额外 interrupt（可能编辑审核打回） payload = {_summarize_event(payload)}") + print("尝试强制定稿…") + payload = await _stream_until_interrupt( + client, thread_id, + command={"resume": {"action": "force_finalize", "overrideVerdict": True}}, + ) + if payload: + print("WARN: 强制定稿后仍有 interrupt，graph 路由可能有意外分支。") + + # ── Step 6: 验证 state 持久化 ──────────────────────────────────────────── + _banner("Step 6: 重新建客户端 client.threads.get_state(thread_id) 应能拿到最终 state") + client2 = get_client(url=DEFAULT_URL) + state = await client2.threads.get_state(thread_id) + values = state.get("values") or {} + final_status = values.get("status") + has_draft = bool(values.get("current_draft")) + print(f"final status = {final_status}, has_draft = {has_draft}, " + f"next nodes = {state.get('next')}") + if final_status not in {"done", "writing_draft", "revising"}: + print("WARN: 最终 status 不是 done，请人工确认上面流程是否正常。") + print("\nPROBE OK —— 阶段 0 验证通过的核心条件已成立（interrupt + resume + state 持久化）。") + return 0 + + +if __name__ == "__main__": + sys.exit(asyncio.run(main()))