Files
deerflow-code/frontend-web/src/open-canvas/contexts/AIWritingContext.tsx
T
fengxiangdebaoziandClaude Opus 4.8 b8c78ba06e feat(ai-writing): 大纲先行确认 + 新文章类型识别 + 参考文献 + 编辑器修复
- 写作配置 Q&A 改为「先流式生成大纲并多轮确认 → 再确认意图」,放宽轮数上限
- 意图识别出列表外的新文章类型时带上类型名+简介(仅本次写作临时使用)
- 成稿支持参考文献:正文 [n] 角标 + 文末按出现顺序的「参考文献」列表(表单开关,默认开)
- 流式接取未闭合 <think> 思考内容,首 token 前加「正在整理…」占位
- 修复右侧编辑器输入空格只出一次、光标跳字(避免自激回灌 setContent 重置光标)

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-23 19:07:03 +08:00

449 lines
18 KiB
TypeScript

import {
createContext,
memo,
useCallback,
useContext,
useEffect,
useMemo,
useReducer,
useRef,
type Dispatch,
type MutableRefObject,
type ReactNode,
type RefObject,
} from 'react';
import { useStream } from '@langchain/langgraph-sdk/react';
import { getAPIClient } from '@/core/api';
import {
getStreamErrorMessage,
isThreadNotAccessibleError,
notifyThreadAccessChangedAndReload,
} from '@/core/threads/thread-access-error';
import { useCanvasGraphContext } from '../contexts/CanvasGraphContext';
import { saveAIWritingTranscript } from '../api/ai-writing-sessions';
import { countArticleWords } from '../utils/wordCount';
import {
INITIAL_STATE,
customToProgressEvent,
interruptToEvent,
nodeUpdatesToEvents,
reducer,
type Action,
type AIWritingValues,
} from '../hooks/useAIWritingStream';
import type {
AIWritingRequest,
AIWritingState,
CompletedIntervention,
ProgressEvent,
UserInterventionPayload,
WritingStatus,
} from '../types/ai-writing';
/** 高频 chunk 类型:合并到同一帧再 dispatch,避免同步嵌套 setState 触发 Maximum update depth。 */
const BATCHED_CHUNK_TYPES = new Set<ProgressEvent['type']>([
'draft_chunk',
'outline_chunk',
'intent_chunk',
'skill_select_chunk',
'review_chunk',
]);
function createProgressEventBatcher(dispatchRef: RefObject<Dispatch<Action>>) {
const pending: ProgressEvent[] = [];
let rafId = 0;
const flush = () => {
rafId = 0;
if (pending.length === 0) return;
const events = pending.splice(0, pending.length);
if (events.length === 1) {
dispatchRef.current?.({ type: 'handle', event: events[0] });
} else {
dispatchRef.current?.({ type: 'handle_batch', events });
}
};
return {
push(event: ProgressEvent) {
pending.push(event);
if (!rafId) rafId = requestAnimationFrame(flush);
},
pushImmediate(event: ProgressEvent) {
// 关键:先把缓冲里的 chunk 同步 flush,再派发立即事件,保证事件顺序。
// 否则 *_ready / await_user / step_* 会插队到尚在 rAF 缓冲里的 *_chunk 之前——
// ready 先清空流式文本 / 切状态,随后残留 chunk 又被回灌,导致:
// 意图/大纲/审核的流式视图清不掉(残留 JSON 尾巴)、切不回结果卡、右侧 markdown 切不回编辑器。
if (rafId) cancelAnimationFrame(rafId);
flush();
dispatchRef.current?.({ type: 'handle', event });
},
cancel() {
if (rafId) cancelAnimationFrame(rafId);
rafId = 0;
pending.length = 0;
},
};
}
type AIWritingStreamApi = Pick<
ReturnType<typeof useStream<AIWritingValues>>,
'submit' | 'stop' | 'interrupt'
>;
export type AIWritingApi = {
state: AIWritingState;
startWriting: (request: AIWritingRequest) => Promise<void>;
submitIntervention: (payload: UserInterventionPayload) => Promise<void>;
pauseWriting: () => void;
resumeWriting: () => void;
reset: () => void;
};
const AIWritingContext = createContext<AIWritingApi | null>(null);
/** 可选读取:不在 Provider 内时返回 null(如 Canvas 布局右侧 artifact 面板)。 */
export function useAIWritingOptional(): AIWritingApi | null {
return useContext(AIWritingContext);
}
/** 在 ``AIWritingProvider`` 内读取写作状态与操作 API。 */
export function useAIWriting(): AIWritingApi {
const ctx = useContext(AIWritingContext);
if (!ctx) {
throw new Error('useAIWriting must be used within AIWritingProvider');
}
return ctx;
}
/**
* 承载 ``useStream`` 的独立组件(render null)。
*
* 用 ``memo`` 且仅随 ``threadId`` 重渲染,这样 ``draft_chunk`` 等高频 custom 事件
* 更新 reducer 时不会重跑 ``useStream``,避免 onCustomEvent → dispatch → 重渲染
* → useStream 重入 的更新风暴(Maximum update depth exceeded)。
*/
const AIWritingStreamHost = memo(function AIWritingStreamHost({
threadId,
dispatchRef,
sessionIdRef,
streamRef,
seenInterruptRef,
}: {
threadId: string | undefined;
dispatchRef: RefObject<Dispatch<Action>>;
sessionIdRef: RefObject<string | null>;
streamRef: MutableRefObject<AIWritingStreamApi | null>;
seenInterruptRef: MutableRefObject<string | null>;
}) {
const eventBatcherRef = useRef<ReturnType<typeof createProgressEventBatcher> | null>(null);
if (!eventBatcherRef.current) {
eventBatcherRef.current = createProgressEventBatcher(dispatchRef);
}
const emitProgressEvent = useCallback((event: ProgressEvent) => {
if (BATCHED_CHUNK_TYPES.has(event.type)) {
eventBatcherRef.current!.push(event);
} else {
eventBatcherRef.current!.pushImmediate(event);
}
}, []);
const stream = useStream<AIWritingValues>({
client: getAPIClient(),
assistantId: 'ai_writing',
threadId,
reconnectOnMount: true,
onCreated(meta) {
dispatchRef.current?.({ type: 'set_session', sessionId: meta.thread_id });
},
onCustomEvent(event: unknown) {
if (typeof window !== 'undefined' && (window as { __AI_WRITING_DEBUG__?: boolean }).__AI_WRITING_DEBUG__) {
const t = (event as { type?: string } | null)?.type;
// eslint-disable-next-line no-console
console.debug(`[ai-writing:custom] +${Math.round(performance.now())}ms type=${t ?? '?'}`, event);
}
const pe = customToProgressEvent(event, sessionIdRef.current ?? '');
if (pe) emitProgressEvent(pe);
},
onUpdateEvent(data: Record<string, unknown> | undefined) {
if (!data) return;
if (typeof window !== 'undefined' && (window as { __AI_WRITING_DEBUG__?: boolean }).__AI_WRITING_DEBUG__) {
// eslint-disable-next-line no-console
console.debug(`[ai-writing:updates] +${Math.round(performance.now())}ms nodes=${Object.keys(data).join(',')}`);
}
if (Array.isArray((data as Record<string, unknown>).__interrupt__)) {
const interrupts = (data as { __interrupt__: unknown[] }).__interrupt__;
for (const intr of interrupts) {
const value = (intr && typeof intr === 'object' && 'value' in intr)
? (intr as { value: unknown }).value : intr;
const ev = interruptToEvent(value, sessionIdRef.current ?? '');
if (ev) emitProgressEvent(ev);
}
return;
}
for (const nodeOutput of Object.values(data)) {
const events = nodeUpdatesToEvents(nodeOutput, sessionIdRef.current ?? '');
for (const ev of events) emitProgressEvent(ev);
}
},
onError(err: unknown) {
eventBatcherRef.current?.cancel();
const message = getStreamErrorMessage(err);
// A "thread not accessible/not found" 404 almost always means the writing
// thread belongs to a previous login session — the user switched accounts
// (or this tab outlived a re-login), so the current account legitimately
// can't access the old thread. Instead of dumping the raw HTTP 404 into
// the panel, re-frame it around the current identity and reload — same UX
// the chat stream gives (see core/threads/hooks.ts).
if (isThreadNotAccessibleError(message)) {
notifyThreadAccessChangedAndReload();
return;
}
dispatchRef.current?.({ type: 'set_error', message });
},
onFinish(finishState) {
eventBatcherRef.current?.cancel();
const finalStatus = (finishState.values as AIWritingValues | undefined)?.status;
if (finalStatus === 'done' || finalStatus === 'error') {
dispatchRef.current?.({
type: 'handle',
event: {
type: finalStatus === 'error' ? 'error' : 'done',
sessionId: finishState.values?.sessionId as string ?? '',
message: finalStatus === 'error' ? '写作出错' : '写作完成',
},
});
}
},
});
streamRef.current = stream;
useEffect(() => {
const intr = stream.interrupt;
if (!intr) {
seenInterruptRef.current = null;
return;
}
const key = intr.id ?? JSON.stringify(intr.value);
if (seenInterruptRef.current === key) return;
seenInterruptRef.current = key;
const ev = interruptToEvent(intr.value, sessionIdRef.current ?? '');
if (ev) emitProgressEvent(ev);
}, [stream.interrupt, emitProgressEvent, seenInterruptRef, sessionIdRef]);
useEffect(() => () => { eventBatcherRef.current?.cancel(); }, []);
return null;
}, (prev, next) => prev.threadId === next.threadId);
export function AIWritingProvider({ children }: { children: ReactNode }) {
const { setArtifact } = useCanvasGraphContext();
const [state, dispatch] = useReducer(reducer, INITIAL_STATE);
const dispatchRef = useRef(dispatch);
dispatchRef.current = dispatch;
const sessionIdRef = useRef<string | null>(state.sessionId);
sessionIdRef.current = state.sessionId;
const streamRef = useRef<AIWritingStreamApi | null>(null);
const seenInterruptRef = useRef<string | null>(null);
const threadId = state.sessionId ?? undefined;
useEffect(() => {
const s = state.status;
const savable = s === 'done' || s === 'error' || s.startsWith('awaiting_');
if (!state.sessionId || !savable || state.progressEvents.length === 0) return;
saveAIWritingTranscript(state.sessionId, {
progressEvents: state.progressEvents,
completedInterventions: state.completedInterventions,
}).catch(() => { });
// eslint-disable-next-line react-hooks/exhaustive-deps
}, [state.status, state.sessionId]);
const startWriting = useCallback(async (request: AIWritingRequest) => {
if (state.status !== 'idle' && state.status !== 'done' && state.status !== 'error') return;
dispatch({ type: 'start', request });
const stream = streamRef.current;
if (!stream) return;
try {
// 所见即所得:用户在表单选了什么 materialSource,前端就原样传过去。
// 两个合法值:general(通用检索,按素材收集专家配置的技能编排)/
// notebook(我的空间);旧值 knowledge_base / skill 后端按 general 兼容。
// 与 material_source 独立的 notebook_search_space_id / notebook_source_ids
// 控制是否真的去笔记后端拉素材(仅当用户选了"我的空间"模式时由表单填充)。
await stream.submit(
{
user_intent: request.userIntent,
// 用户自带大纲原文(留空则后端走自动规划)
user_outline_raw: request.userOutline?.trim() || '',
article_type: request.articleType,
// 新(列表外)文章类型的一句话特征,仅本次写作临时使用;命中已有类型时为空。
article_type_detail: request.articleTypeDetail?.trim() || '',
target_audience: request.targetAudience,
word_count_target: request.wordCountTarget,
keyword_count: Math.max(1, Math.min(10, request.keywordCount)),
writing_mode: request.writingMode === 'strict' ? 'strict' : 'loose',
// 是否加参考文献(正文 [n] + 文末「参考文献」列表);缺省 false 不改其它入口行为。
add_references: request.addReferences ?? false,
material_source: request.materialSource,
notebook_search_space_id: request.noteSearchSpaceId ?? null,
notebook_source_ids: request.noteSourceIds ?? null,
revision_count: 0,
max_revisions: 3,
status: 'researching',
drafts: [],
progress_events: [],
material_package: null,
current_outline: null,
current_draft: null,
review_result: null,
last_user_intervention: null,
} as Partial<AIWritingValues>,
{
config: { configurable: { model_name: request.modelName ?? '' } } as Record<string, unknown>,
streamMode: ['values', 'updates', 'custom', 'messages-tuple'],
},
);
} catch (err) {
const message = getStreamErrorMessage(err);
if (isThreadNotAccessibleError(message)) {
notifyThreadAccessChangedAndReload();
return;
}
dispatch({ type: 'set_error', message });
}
}, [state.status]);
const submitIntervention = useCallback(async (payload: UserInterventionPayload) => {
if (!state.sessionId || !state.currentPause) return;
const stream = streamRef.current;
if (!stream) return;
const pausePoint = state.currentPause.pausePoint;
const record: CompletedIntervention = {
pausePoint: pausePoint as CompletedIntervention['pausePoint'],
action: payload.action,
};
if (pausePoint === 'material_confirm') record.materials = state.materials;
else if (pausePoint === 'outline_confirm') record.outline = state.outline ?? undefined;
else if (pausePoint === 'draft_confirm') {
record.draftTitle = state.currentDraftTitle;
record.draftWordCount = countArticleWords(state.currentDraftMarkdown);
record.reviewResult = state.reviewResult ?? undefined;
} else if (pausePoint === 'review_confirm') {
record.reviewResult = state.reviewResult ?? undefined;
}
let pendingRevisionContext: AIWritingState['pendingRevisionContext'] = null;
const isApplyingReview = payload.selectedIssueIndices !== undefined && state.reviewResult != null;
if (isApplyingReview && state.reviewResult) {
const allIssues = state.reviewResult.revisionNotes;
const selectedSet = new Set(payload.selectedIssueIndices!);
pendingRevisionContext = {
submittedIssues: allIssues.filter((_, i) => selectedSet.has(i)),
skippedIssues: allIssues.filter((_, i) => !selectedSet.has(i)),
oldWordCount: countArticleWords(state.currentDraftMarkdown),
};
}
let nextStatus: WritingStatus = state.status;
if (pausePoint === 'material_confirm') {
nextStatus = (payload.action === 're_search' || payload.action === 'enable_skill_search')
? 'researching' : 'writing_outline';
} else if (pausePoint === 'outline_confirm') {
nextStatus = payload.action === 're_outline' ? 'writing_outline' : 'writing_draft';
} else if (pausePoint === 'draft_confirm') {
if (payload.action === 'finalize') nextStatus = 'done';
else if (payload.action === 'to_editor') nextStatus = 'reviewing';
else nextStatus = 'revising';
} else if (pausePoint === 'review_confirm') {
nextStatus = payload.action === 'force_finalize' ? 'done' : 'revising';
} else if (pausePoint === 'section_help') {
nextStatus = payload.sectionDecisions?.some((d) => d.decision === 'supplement')
? 'researching' : 'writing_draft';
}
const resumePayload: Record<string, unknown> = { action: payload.action };
if (payload.approvedMaterialIds !== undefined) resumePayload.approvedMaterialIds = payload.approvedMaterialIds;
if (payload.extraKeywords !== undefined) resumePayload.extraKeywords = payload.extraKeywords;
if (payload.userQuery !== undefined) resumePayload.userQuery = payload.userQuery;
if (payload.editedOutline !== undefined) resumePayload.editedOutline = payload.editedOutline;
if (payload.outlineFeedback !== undefined) resumePayload.outlineFeedback = payload.outlineFeedback;
if (payload.userRevisionNotes !== undefined) resumePayload.userRevisionNotes = payload.userRevisionNotes;
if (payload.overrideVerdict !== undefined) resumePayload.overrideVerdict = payload.overrideVerdict;
if (payload.sectionDecisions !== undefined) {
resumePayload.sectionDecisions = payload.sectionDecisions.map((d) => ({
sectionTitle: d.sectionTitle, decision: d.decision,
}));
}
const intr = stream.interrupt;
if (intr) {
seenInterruptRef.current = intr.id ?? JSON.stringify(intr.value);
}
dispatch({ type: 'intervention_submitted', record, nextStatus, pendingRevisionContext });
try {
// resume 是新 invocation,其 config 默认不带最初的 model_name;而写作/审核节点
// 在 resume 后才运行,若首轮 intent 未把模型存进 state(历史会话 / 捕获为空),
// 后端就只能回退默认模型。这里把表单选定的模型随每次 resume 一并回传,
// 让后端(get_writing_model / resolve_agent_model_name 的 configurable 兜底)
// 始终用上用户真正选的模型。
const resumeModelName = state.request?.modelName;
await stream.submit(undefined, {
command: { resume: resumePayload },
streamMode: ['values', 'updates', 'custom', 'messages-tuple'],
...(resumeModelName ? { config: { configurable: { model_name: resumeModelName } } } : {}),
});
} catch (err) {
const message = getStreamErrorMessage(err);
if (isThreadNotAccessibleError(message)) {
notifyThreadAccessChangedAndReload();
return;
}
dispatch({ type: 'set_error', message });
}
}, [state.sessionId, state.currentPause, state.materials, state.outline,
state.currentDraftTitle, state.currentDraftMarkdown, state.reviewResult,
state.status]);
const pauseWriting = useCallback(() => {
void streamRef.current?.stop();
dispatch({ type: 'pause' });
}, []);
const resumeWriting = useCallback(() => {
dispatch({ type: 'resume_ui' });
}, []);
const reset = useCallback(() => {
void streamRef.current?.stop();
setArtifact(null);
seenInterruptRef.current = null;
dispatch({ type: 'reset' });
}, [setArtifact]);
const value = useMemo(
() => ({ state, startWriting, submitIntervention, pauseWriting, resumeWriting, reset }),
[state, startWriting, submitIntervention, pauseWriting, resumeWriting, reset],
);
return (
<AIWritingContext.Provider value={value}>
<AIWritingStreamHost
threadId={threadId}
dispatchRef={dispatchRef}
sessionIdRef={sessionIdRef}
streamRef={streamRef}
seenInterruptRef={seenInterruptRef}
/>
{children}
</AIWritingContext.Provider>
);
}