fix: improve ai writing chat streaming and intent handling

offline-backend-20260512/backend/app/gateway/routers/ai_writing.py

@@ -23,10 +23,12 @@ import tempfile
 from pathlib import Path
 from typing import Optional
 
-from fastapi import APIRouter, File, HTTPException, Request, UploadFile
+from fastapi import APIRouter, Depends, File, HTTPException, Request, UploadFile
+from fastapi.responses import StreamingResponse
 from pydantic import BaseModel, Field
 
 from app.gateway.authz import require_auth
+from app.gateway.deps import get_config
 from deerflow.utils.file_conversion import CONVERTIBLE_EXTENSIONS, convert_file_to_markdown
 
 logger = logging.getLogger(__name__)
@@ -353,30 +355,53 @@ class IntentParseResponse(BaseModel):
     clarify: str = ""
 
 
-@router.post("/sessions/{session_id}/intent", response_model=IntentParseResponse)
-async def parse_session_intent(session_id: str, body: IntentParseRequest, request: Request):
-    """对话驱动写作台：把用户自由文本翻译成结构化写作指令（或问答/澄清）。
+class WritingSuggestionMessage(BaseModel):
+    role: str
+    content: str
 
-    读该会话 graph checkpoint 的素材/大纲/草稿/审核摘要 + 当前 pending interrupt
-    （权威暂停点），交给 ``deerflow.agents.ai_writing.intent_router`` 的 LLM 意图
-    路由。动作白名单 / 高代价置信度门槛 / payload 清洗都在 intent_router 内完成，
-    这里只做鉴权与 state 读取。模型失败不抛 5xx —— intent_router 会降级成 clarify。
-    """
-    from app.gateway.ai_writing_job_executor import _pending_interrupt
-    from app.gateway.deps import get_ai_writing_session_store, get_current_user
-    from deerflow.agents.ai_writing.intent_router import parse_user_intent
 
+class WritingSuggestionsRequest(BaseModel):
+    messages: list[WritingSuggestionMessage] = Field(default_factory=list)
+    n: int = Field(default=3, ge=1, le=5)
+    model_name: Optional[str] = None
+
+
+class WritingSuggestionsResponse(BaseModel):
+    suggestions: list[str] = Field(default_factory=list)
+
+
+class AnswerStreamRequest(BaseModel):
+    text: str
+    status: Optional[str] = None
+    pause_point: Optional[str] = None
+    model_name: Optional[str] = None
+
+
+async def _prepare_intent_parse_context(
+    session_id: str,
+    body: IntentParseRequest,
+    request: Request,
+) -> tuple[str, dict, Optional[dict], Optional[str]]:
+    """鉴权并读取会话 checkpoint，为一次性/流式意图解析提供同一份上下文。"""
     text = (body.text or "").strip()
     if not text:
         raise HTTPException(status_code=400, detail="text 不能为空")
 
-    repo = get_ai_writing_session_store(request)
-    if repo is None:
-        raise HTTPException(status_code=503, detail="Session store not available")
-    user_id = await get_current_user(request)
-    record = await repo.get(session_id, user_id=user_id)
-    if record is None:
-        raise HTTPException(status_code=404, detail="Session not found")
+    await _ensure_session_owner(session_id, request)
+    values, interrupt_value, pause_point = await _read_ai_writing_checkpoint_context(
+        session_id,
+        body.pause_point,
+        request,
+    )
+    return text, values, interrupt_value, pause_point
+
+
+async def _read_ai_writing_checkpoint_context(
+    session_id: str,
+    fallback_pause_point: Optional[str],
+    request: Request,
+) -> tuple[dict, Optional[dict], Optional[str]]:
+    from app.gateway.ai_writing_job_executor import _pending_interrupt
 
     # 读 checkpoint：暂停点以 pending interrupt 为权威（frontend 声明只作读取失败兜底）。
     values: dict = {}
@@ -400,7 +425,89 @@ async def parse_session_intent(session_id: str, body: IntentParseRequest, reques
         except Exception:
             logger.warning("ai_writing intent: read checkpoint failed for %s", session_id, exc_info=True)
     if pause_point is None and not checkpoint_ok:
-        pause_point = (body.pause_point or "").strip() or None
+        pause_point = (fallback_pause_point or "").strip() or None
+
+    return values, interrupt_value, pause_point
+
+
+async def _get_session_owner_record(session_id: str, request: Request) -> dict:
+    from app.gateway.deps import get_ai_writing_session_store, get_current_user
+
+    repo = get_ai_writing_session_store(request)
+    if repo is None:
+        raise HTTPException(status_code=503, detail="Session store not available")
+    user_id = await get_current_user(request)
+    record = await repo.get(session_id, user_id=user_id)
+    if record is None:
+        raise HTTPException(status_code=404, detail="Session not found")
+    return record
+
+
+async def _ensure_session_owner(session_id: str, request: Request) -> None:
+    await _get_session_owner_record(session_id, request)
+
+
+def _clip_answer_context(text: str, limit: int) -> str:
+    text = (text or "").strip()
+    if len(text) <= limit:
+        return text
+    return text[:limit] + "\n…（内容过长已截断）"
+
+
+def _build_session_answer_context(record: dict) -> str:
+    """Build a fast context from the persisted session row for plain Q&A."""
+    parts: list[str] = []
+    title = str(record.get("title") or record.get("draft_title") or "").strip()
+    user_intent = str(record.get("user_intent") or "").strip()
+    status = str(record.get("status") or "").strip()
+    if title:
+        parts.append(f"会话标题：{title}")
+    if user_intent:
+        parts.append(f"写作主题：{_clip_answer_context(user_intent, 300)}")
+    if status:
+        parts.append(f"流程状态：{status}")
+
+    draft_markdown = str(record.get("draft_markdown") or "").strip()
+    has_substantive_context = bool(draft_markdown)
+    if draft_markdown:
+        draft_title = str(record.get("draft_title") or title or "").strip()
+        header = f"草稿标题：《{draft_title}》" if draft_title else "当前草稿"
+        parts.append(f"{header}\n--- 草稿正文 ---\n{_clip_answer_context(draft_markdown, 6000)}")
+
+    review = record.get("review_result") if isinstance(record.get("review_result"), dict) else {}
+    if review:
+        has_substantive_context = True
+        lines = [
+            f"编辑结论：{review.get('verdict', '')}（总分 {review.get('overall_score', review.get('overallScore', ''))}）"
+        ]
+        notes = review.get("revision_notes") or review.get("revisionNotes") or []
+        if isinstance(notes, list):
+            for i, note in enumerate(notes[:8]):
+                if not isinstance(note, dict):
+                    continue
+                issue = _clip_answer_context(str(note.get("issue") or ""), 120)
+                suggestion = _clip_answer_context(str(note.get("suggestion") or ""), 120)
+                if issue or suggestion:
+                    lines.append(f"意见 {i + 1}：{issue} → {suggestion}")
+        parts.append("\n".join(lines))
+
+    if not has_substantive_context:
+        return ""
+    return "\n\n".join(part for part in parts if part.strip())
+
+
+@router.post("/sessions/{session_id}/intent", response_model=IntentParseResponse)
+async def parse_session_intent(session_id: str, body: IntentParseRequest, request: Request):
+    """对话驱动写作台：把用户自由文本翻译成结构化写作指令（或问答/澄清）。
+
+    读该会话 graph checkpoint 的素材/大纲/草稿/审核摘要 + 当前 pending interrupt
+    （权威暂停点），交给 ``deerflow.agents.ai_writing.intent_router`` 的 LLM 意图
+    路由。动作白名单 / 高代价置信度门槛 / payload 清洗都在 intent_router 内完成，
+    这里只做鉴权与 state 读取。模型失败不抛 5xx —— intent_router 会降级成 clarify。
+    """
+    from deerflow.agents.ai_writing.intent_router import parse_user_intent
+
+    text, values, interrupt_value, pause_point = await _prepare_intent_parse_context(session_id, body, request)
 
     result = await parse_user_intent(
         text=text,
@@ -412,6 +519,231 @@ async def parse_session_intent(session_id: str, body: IntentParseRequest, reques
     return IntentParseResponse(**result)
 
 
+@router.post("/sessions/{session_id}/intent/stream")
+async def stream_session_intent(session_id: str, body: IntentParseRequest, request: Request):
+    """对话驱动写作台：流式意图解析。
+
+    SSE 事件：
+      - ``thinking``: ``{"chunk": "..."}``，模型推理 / ``<think>`` 内容
+      - ``text``: ``{"chunk": "..."}``，模型正文输出（通常是结构化 JSON）
+      - ``result``: 最终规范化后的 ``IntentParseResponse`` 字段
+      - ``done`` / ``error``
+    """
+    from app.gateway.services import format_sse
+    from deerflow.agents.ai_writing.intent_router import stream_user_intent
+
+    text, values, interrupt_value, pause_point = await _prepare_intent_parse_context(session_id, body, request)
+
+    async def event_stream():
+        try:
+            async for event in stream_user_intent(
+                text=text,
+                pause_point=pause_point,
+                values=values,
+                interrupt_value=interrupt_value,
+                model_name=str(values.get("model_name") or "") or None,
+            ):
+                if event.get("type") == "chunk":
+                    chunk = str(event.get("chunk") or "")
+                    if not chunk:
+                        continue
+                    yield format_sse(
+                        "thinking" if event.get("is_thinking") else "text",
+                        {"chunk": chunk},
+                    )
+                elif event.get("type") == "result":
+                    yield format_sse("result", event.get("result") or {})
+            yield format_sse("done", {})
+        except Exception:
+            logger.exception("ai_writing intent stream failed: session=%s", session_id)
+            yield format_sse("error", {"message": "意图解析流式接口异常，请稍后重试"})
+
+    return StreamingResponse(
+        event_stream(),
+        media_type="text/event-stream",
+        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
+    )
+
+
+@router.post("/sessions/{session_id}/suggestions", response_model=WritingSuggestionsResponse)
+async def generate_session_suggestions(
+    session_id: str,
+    body: WritingSuggestionsRequest,
+    request: Request,
+    config=Depends(get_config),
+):
+    """为 AI 写作对话版生成窄栏追问建议。
+
+    与主对话页的 ``/api/threads/{id}/suggestions`` 语义一致，但鉴权对象换成
+    AI 写作 session，避免把写作 sessionId 冒充 workspace threadId。
+    """
+    from langchain_core.messages import HumanMessage, SystemMessage
+
+    from app.gateway.routers.suggestions import (
+        _extract_response_text,
+        _format_conversation,
+        _parse_json_string_list,
+        SuggestionMessage,
+    )
+    from deerflow.models import create_chat_model
+
+    await _ensure_session_owner(session_id, request)
+
+    if not body.messages:
+        return WritingSuggestionsResponse()
+
+    messages = [
+        SuggestionMessage(role=m.role, content=m.content)
+        for m in body.messages
+        if (m.content or "").strip()
+    ]
+    conversation = _format_conversation(messages)
+    if not conversation:
+        return WritingSuggestionsResponse()
+
+    n = body.n
+    system_instruction = (
+        "You are generating compact follow-up questions for a narrow AI writing sidebar.\n"
+        f"Based on the conversation below, produce EXACTLY {n} short questions the user might ask next.\n"
+        "Requirements:\n"
+        "- Questions must be relevant to the article or writing workflow.\n"
+        "- Prefer questions about summary, structure, evidence, style, revisions, or next actions.\n"
+        "- Questions must be written in the same language as the user.\n"
+        "- Keep each question very concise (<= 24 Chinese characters when Chinese).\n"
+        "- Do NOT include numbering, markdown, or any extra text.\n"
+        "- Output MUST be a JSON array of strings only.\n"
+    )
+    user_content = f"Conversation Context:\n{conversation}\n\nGenerate {n} follow-up questions"
+
+    try:
+        model = create_chat_model(name=body.model_name, thinking_enabled=False, app_config=config)
+        response = await model.ainvoke(
+            [SystemMessage(content=system_instruction), HumanMessage(content=user_content)],
+            config={"run_name": "ai_writing_suggestions"},
+        )
+        raw = _extract_response_text(response.content)
+        suggestions = _parse_json_string_list(raw) or []
+        cleaned = [s.replace("\n", " ").strip() for s in suggestions if s.strip()]
+        return WritingSuggestionsResponse(suggestions=cleaned[:n])
+    except Exception:
+        logger.exception("ai_writing: failed to generate suggestions: session=%s", session_id)
+        return WritingSuggestionsResponse()
+
+
+@router.post("/sessions/{session_id}/answer/stream")
+async def stream_session_answer(session_id: str, body: AnswerStreamRequest, request: Request):
+    """AI 写作对话版问答旁路：纯文本 SSE 回答。
+
+    意图路由仍负责操作类消息；明显的提问走这里，避免把答案塞进 JSON 的
+    ``answer`` 字段导致“模型完整生成后才一次性显示”。
+    """
+    from app.gateway.services import format_sse
+    from deerflow.agents.ai_writing.intent_router import build_context_snapshot
+    from deerflow.agents.ai_writing.nodes._utils import ThinkTagParser, _content_to_text, reasoning_delta
+    from deerflow.models import create_chat_model
+    from langchain_core.messages import HumanMessage, SystemMessage
+
+    text = (body.text or "").strip()
+    if not text:
+        raise HTTPException(status_code=400, detail="text 不能为空")
+    record = await _get_session_owner_record(session_id, request)
+    session_context = _build_session_answer_context(record)
+
+    async def event_stream():
+        import time
+
+        started = time.monotonic()
+        parser = ThinkTagParser()
+        chunk_count = 0
+        first_chunk_logged = False
+        yield format_sse("start", {})
+        try:
+            values: dict = {}
+            if session_context:
+                context = session_context
+                model_name = (body.model_name or "").strip() or None
+                logger.info(
+                    "ai_writing answer context ready: session=%s elapsed=%.3fs source=session_row pause_point=%s",
+                    session_id,
+                    time.monotonic() - started,
+                    body.pause_point,
+                )
+            else:
+                values, interrupt_value, pause_point = await _read_ai_writing_checkpoint_context(
+                    session_id,
+                    body.pause_point,
+                    request,
+                )
+                context = build_context_snapshot(values, pause_point, interrupt_value)
+                model_name = (body.model_name or "").strip() or str(values.get("model_name") or "") or None
+                logger.info(
+                    "ai_writing answer context ready: session=%s elapsed=%.3fs source=checkpoint has_values=%s pause_point=%s",
+                    session_id,
+                    time.monotonic() - started,
+                    bool(values),
+                    pause_point,
+                )
+
+            messages = [
+                SystemMessage(content=(
+                    "你是 AI 智能写作工作台里的问答助手。用户正在写一篇文章，"
+                    "你只能根据给定的写作上下文回答问题，不执行确认、检索、修改、定稿等流程动作。"
+                    "如果上下文没有足够信息，请明确说明，不要编造。回答用中文，简洁但有信息量。"
+                )),
+                HumanMessage(content=f"【写作上下文】\n{context}\n\n【用户问题】\n{text}\n\n请直接回答："),
+            ]
+
+            model = create_chat_model(
+                name=model_name,
+                thinking_enabled=False,
+                force_disable_thinking=True,
+            )
+            async for chunk in model.astream(messages, config={"run_name": "ai_writing_answer"}):
+                reasoning = reasoning_delta(chunk)
+                if reasoning:
+                    if not first_chunk_logged:
+                        first_chunk_logged = True
+                        logger.info(
+                            "ai_writing answer first chunk: session=%s elapsed=%.3fs kind=thinking",
+                            session_id,
+                            time.monotonic() - started,
+                        )
+                    yield format_sse("thinking", {"chunk": reasoning})
+                content = _content_to_text(getattr(chunk, "content", "") or "")
+                if not content:
+                    continue
+                for is_thinking, part in parser.feed(content):
+                    if not part:
+                        continue
+                    if not first_chunk_logged:
+                        first_chunk_logged = True
+                        logger.info(
+                            "ai_writing answer first chunk: session=%s elapsed=%.3fs kind=%s",
+                            session_id,
+                            time.monotonic() - started,
+                            "thinking" if is_thinking else "text",
+                        )
+                    chunk_count += 1
+                    yield format_sse("thinking" if is_thinking else "text", {"chunk": part})
+            logger.info(
+                "ai_writing answer stream done: session=%s chunks=%d elapsed=%.3fs first_chunk=%s",
+                session_id,
+                chunk_count,
+                time.monotonic() - started,
+                first_chunk_logged,
+            )
+            yield format_sse("done", {})
+        except Exception:
+            logger.exception("ai_writing answer stream failed: session=%s", session_id)
+            yield format_sse("error", {"message": "回答生成失败，请稍后重试"})
+
+    return StreamingResponse(
+        event_stream(),
+        media_type="text/event-stream",
+        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
+    )
+
+
 class BackgroundResponse(BaseModel):
     # 是否新拉起了后台任务（False = 已在后台跑，幂等返回）
     started: bool

offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/intent_router.py

@@ -17,7 +17,7 @@ from __future__ import annotations
 
 import logging
 import re
-from typing import Any
+from typing import Any, AsyncIterator
 
 from .prompts.intent_router_prompts import (
     INTENT_ROUTER_SYSTEM,
@@ -48,6 +48,73 @@ ACTION_CONFIDENCE = 0.5
 _SECTION_DECISIONS = {"supplement", "loose", "delete"}
 
 _THINK_RE = re.compile(r"<think>.*?</think>", re.S)
+_WRITING_ACTION_VERB_RE = re.compile(
+    r"(新增|增加|添加|加入|加上|补充|补上|补个|补一个|生成|撰写|写一|写个|写成|扩写|缩写|精简|改写|润色|调整|修改|重写|删除|删掉|拆成|合并|挪到|移动|规划|重新规划|设计|安排)"
+)
+_WRITING_ACTION_TARGET_RE = re.compile(
+    r"(第[一二三四五六七八九十\d]+[章节段]|章节|章|小节|段落|段|结尾|开头|标题|大纲|内容|案例|表格|总结章|总结章节|结论|摘要|观点|部分)"
+)
+_INFO_QUESTION_RE = re.compile(r"(哪些|什么|为什么|是什么|有哪些|多少|是否)")
+
+
+def _fallback_intent() -> dict:
+    return {
+        "intent": "clarify",
+        "action": None,
+        "payload": {},
+        "need_research": False,
+        "research_query": "",
+        "answer": "",
+        "confidence": 0.0,
+        "clarify": "我没有理解你的意思，能换个说法再试一次吗？也可以直接点击上方卡片里的按钮操作。",
+    }
+
+
+def _looks_like_writing_change_request(text: str) -> bool:
+    text = (text or "").strip()
+    if _INFO_QUESTION_RE.search(text):
+        return False
+    return bool(_WRITING_ACTION_VERB_RE.search(text) and _WRITING_ACTION_TARGET_RE.search(text))
+
+
+def _rule_based_intent(text: str, pause_point: str | None) -> dict | None:
+    """Deterministic guard for writing-operation requests that look like Q&A."""
+    if not _looks_like_writing_change_request(text):
+        return None
+    if pause_point == "outline_confirm":
+        return {
+            "intent": "revise_outline",
+            "action": "re_outline",
+            "payload": {"outlineFeedback": text.strip()},
+            "need_research": False,
+            "research_query": "",
+            "answer": "",
+            "confidence": 1.0,
+            "clarify": "",
+        }
+    if pause_point == "draft_confirm":
+        return {
+            "intent": "revise_draft",
+            "action": "user_revise",
+            "payload": {"userRevisionNotes": text.strip()},
+            "need_research": False,
+            "research_query": "",
+            "answer": "",
+            "confidence": 1.0,
+            "clarify": "",
+        }
+    if pause_point == "review_confirm":
+        return {
+            "intent": "accept_review_with_user_notes",
+            "action": "accept_review",
+            "payload": {"userRevisionNotes": text.strip()},
+            "need_research": False,
+            "research_query": "",
+            "answer": "",
+            "confidence": 1.0,
+            "clarify": "",
+        }
+    return None
 
 
 # ── 上下文快照 ────────────────────────────────────────────────────────────────
@@ -362,16 +429,10 @@ async def parse_user_intent(
     """
     from .nodes._utils import _content_to_text, parse_json
 
-    fallback = {
-        "intent": "clarify",
-        "action": None,
-        "payload": {},
-        "need_research": False,
-        "research_query": "",
-        "answer": "",
-        "confidence": 0.0,
-        "clarify": "我没有理解你的意思，能换个说法再试一次吗？也可以直接点击上方卡片里的按钮操作。",
-    }
+    fallback = _fallback_intent()
+    rule = _rule_based_intent(text, pause_point)
+    if rule is not None:
+        return normalize_intent_result(rule, pause_point, values, interrupt_value)
 
     messages = _build_messages(text, pause_point, values, interrupt_value)
 
@@ -407,3 +468,106 @@ async def parse_user_intent(
 
     parsed = parse_json(raw_text, fallback)
     return normalize_intent_result(parsed, pause_point, values, interrupt_value)
+
+
+async def stream_user_intent(
+    *,
+    text: str,
+    pause_point: str | None,
+    values: dict,
+    interrupt_value: dict | None = None,
+    model_name: str | None = None,
+    model: Any | None = None,
+) -> AsyncIterator[dict]:
+    """流式解析用户自由文本。
+
+    事件契约：
+      - ``{"type": "chunk", "chunk": "...", "is_thinking": bool}``
+      - ``{"type": "result", "result": normalized_intent}``
+
+    解析阶段仍以结构化 JSON 为最终产物；流式输出用于前端实时展示模型的
+    思考/解析过程。``<think>...</think>`` 内容和 provider reasoning 字段都会作为
+    ``is_thinking=True`` 分流，正文 JSON 只参与最终解析。
+    """
+    from .nodes._utils import ThinkTagParser, _content_to_text, parse_json, reasoning_delta
+
+    fallback = _fallback_intent()
+    rule = _rule_based_intent(text, pause_point)
+    if rule is not None:
+        yield {
+            "type": "result",
+            "result": normalize_intent_result(rule, pause_point, values, interrupt_value),
+        }
+        return
+
+    messages = _build_messages(text, pause_point, values, interrupt_value)
+
+    def content_events(parser: ThinkTagParser, raw_parts: list[str], delta: str) -> list[dict]:
+        events = []
+        for is_thinking, part in parser.feed(delta):
+            if not is_thinking:
+                raw_parts.append(part)
+            events.append({"type": "chunk", "chunk": part, "is_thinking": is_thinking})
+        return events
+
+    async def run_model(m: Any) -> AsyncIterator[dict]:
+        parser = ThinkTagParser()
+        raw_parts: list[str] = []
+        if hasattr(m, "astream"):
+            async for chunk in m.astream(messages):
+                reasoning = reasoning_delta(chunk)
+                if reasoning:
+                    yield {"type": "chunk", "chunk": reasoning, "is_thinking": True}
+                content = _content_to_text(getattr(chunk, "content", "") or "")
+                if content:
+                    for event in content_events(parser, raw_parts, content):
+                        yield event
+        else:
+            resp = await m.ainvoke(messages)
+            content = _content_to_text(getattr(resp, "content", "") or "")
+            if content:
+                for event in content_events(parser, raw_parts, content):
+                    yield event
+        yield {"type": "_raw_text", "raw_text": "".join(raw_parts)}
+
+    raw_text = ""
+    if model is not None:
+        try:
+            async for event in run_model(model):
+                if event["type"] == "_raw_text":
+                    raw_text = str(event.get("raw_text") or "")
+                else:
+                    yield event
+        except Exception:
+            raw_text = ""
+            logger.exception("intent_router: injected stream model failed")
+    else:
+        from deerflow.models import create_chat_model
+
+        from .nodes._model_fallback import classify_exception, fallback_chain
+
+        chain = fallback_chain((model_name or "").strip() or None) or [None]
+        for candidate in chain:
+            try:
+                m = create_chat_model(name=candidate) if candidate else create_chat_model()
+                async for event in run_model(m):
+                    if event["type"] == "_raw_text":
+                        raw_text = str(event.get("raw_text") or "")
+                    else:
+                        yield event
+                break
+            except Exception as exc:  # noqa: BLE001 — 换下一个模型继续
+                raw_text = ""
+                logger.warning(
+                    "intent_router: stream model=%s failed (reason=%s): %s",
+                    candidate, classify_exception(exc), exc,
+                )
+                continue
+
+    raw_text = _THINK_RE.sub("", raw_text or "").strip()
+    if not raw_text:
+        result = normalize_intent_result(fallback, pause_point, values, interrupt_value)
+    else:
+        parsed = parse_json(raw_text, fallback)
+        result = normalize_intent_result(parsed, pause_point, values, interrupt_value)
+    yield {"type": "result", "result": result}