ai写作

2026-05-17 19:29:10 +08:00
1 parent 6cab62d140
commit 918e65d8d3
@@ -122,6 +122,10 @@ async def langgraph_runtime(app: FastAPI) -> AsyncGenerator[None, None]:

        app.state.tag_store = make_tag_store(sf)

+        from deerflow.persistence.ai_writing_sessions import make_ai_writing_session_store
+
+        app.state.ai_writing_session_store = make_ai_writing_session_store(sf)
+
        # Run event store (has its own factory with config-driven backend selection)
        run_events_config = getattr(config, "run_events", None)
        app.state.run_event_store = make_run_event_store(run_events_config)
@@ -222,6 +226,11 @@ def get_tag_store(request: Request) -> TagStore:
    return val


+def get_ai_writing_session_store(request: Request):
+    """Return the AI writing session store, or None when not configured."""
+    return getattr(request.app.state, "ai_writing_session_store", None)
+
+
 def get_run_context(request: Request) -> RunContext:
    """Build a :class:`RunContext` from ``app.state`` singletons.

@@ -1,9 +1,13 @@
 """AI 写作路由 —— 独立于原有路由，前缀 /api/ai-writing/

-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     查询当前流程状态
+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}        删除会话
 """
 from __future__ import annotations

@@ -12,9 +16,10 @@ 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
+from fastapi import APIRouter, HTTPException, Request
 from fastapi.responses import StreamingResponse
 from langgraph.types import Command
 from pydantic import BaseModel, Field
@@ -40,12 +45,32 @@ class StartWritingResponse(BaseModel):

 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
+
+
+class RenameSessionRequest(BaseModel):
+    title: str
+
+
+class SessionResponse(BaseModel):
+    id: str
+    user_id: Optional[str] = None
+    title: str
+    user_intent: Optional[str] = None
+    status: str
+    draft_title: Optional[str] = None
+    draft_markdown: Optional[str] = None
+    completed_interventions: Optional[list] = None
+    review_result: Optional[dict] = None
+    created_at: str
+    updated_at: str


 # ── 内部状态 ──────────────────────────────────────────────────────────────────
@@ -56,6 +81,18 @@ _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] = {}
+

 # ── 工具函数 ──────────────────────────────────────────────────────────────────

@@ -76,17 +113,41 @@ def _make_config(session_id: str, model_name: str = "") -> dict:
    }


+def _truncate_title(text: str, max_len: int = 50) -> str:
+    text = text.strip()
+    return text[:max_len] + ("…" if len(text) > max_len else "")
+
+
 # ── 路由 ──────────────────────────────────────────────────────────────────────

@router.post("/start", response_model=StartWritingResponse)
-async def start_writing(body: StartWritingRequest):
+async def start_writing(body: StartWritingRequest, request: Request):
    """启动写作流程，返回 session_id。"""
    from deerflow.agents.ai_writing.graph import AI_WRITING_GRAPH
+    from app.gateway.deps import get_ai_writing_session_store, get_current_user

    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] = []
+
+    # 持久化会话记录（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,
@@ -110,6 +171,63 @@ async def start_writing(body: StartWritingRequest):
    return StartWritingResponse(session_id=session_id)


+@router.get("/sessions", response_model=list[SessionResponse])
+async def list_sessions(request: Request):
+    """列出当前用户的 AI 写作历史会话。"""
+    from app.gateway.deps import get_ai_writing_session_store, get_current_user
+
+    repo = get_ai_writing_session_store(request)
+    if repo is None:
+        return []
+    user_id = await get_current_user(request)
+    return await repo.list(user_id=user_id)
+
+
+@router.get("/sessions/{session_id}", response_model=SessionResponse)
+async def get_session(session_id: str, request: Request):
+    """获取单个历史会话。"""
+    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
+
+
+@router.patch("/sessions/{session_id}", response_model=SessionResponse)
+async def rename_session(session_id: str, body: RenameSessionRequest, request: Request):
+    """重命名会话标题。"""
+    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)
+    updated = await repo.update(session_id, user_id=user_id, title=body.title.strip())
+    if updated is None:
+        raise HTTPException(status_code=404, detail="Session not found")
+    return updated
+
+
+@router.delete("/sessions/{session_id}")
+async def delete_session(session_id: str, request: Request):
+    """删除历史会话。"""
+    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)
+    deleted = await repo.delete(session_id, user_id=user_id)
+    if not deleted:
+        raise HTTPException(status_code=404, detail="Session not found")
+    return {"success": True}
+
+
@router.get("/{session_id}/stream")
 async def stream_progress(session_id: str):
    """SSE 流：推送进度事件、干预请求、草稿内容。"""
@@ -121,9 +239,10 @@ async def stream_progress(session_id: str):


@router.post("/{session_id}/resume")
-async def resume_writing(session_id: str, body: ResumeRequest):
+async def resume_writing(session_id: str, body: ResumeRequest, request: Request):
    """用户提交干预决策，恢复暂停的流程。"""
    from deerflow.agents.ai_writing.graph import AI_WRITING_GRAPH
+    from app.gateway.deps import get_current_user

    if session_id not in _event_queues:
        raise HTTPException(status_code=404, detail="Session not found")
@@ -137,8 +256,29 @@ async def resume_writing(session_id: str, body: ResumeRequest):
        "outlineFeedback": body.outline_feedback,
        "userRevisionNotes": body.user_revision_notes,
        "overrideVerdict": body.override_verdict,
+        "selectedIssueIndices": body.selected_issue_indices,
    }

+    # 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)
+
    asyncio.create_task(
        _stream_graph_segment(session_id, Command(resume=intervention), config, AI_WRITING_GRAPH)
    )
@@ -174,6 +314,7 @@ async def _stream_graph_segment(
        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))
@@ -208,18 +349,48 @@ async def _stream_graph_segment(
            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.get("title", ""),
-                    "markdown": draft.get("full_markdown", ""),
+                    "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":
@@ -232,12 +403,30 @@ async def _stream_graph_segment(
    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)


 def _event_type(evt: dict) -> str:
@@ -0,0 +1,11 @@
+from deerflow.persistence.ai_writing_sessions.model import AIWritingSessionRow
+from deerflow.persistence.ai_writing_sessions.sql import AIWritingSessionRepository
+
+
+def make_ai_writing_session_store(session_factory) -> AIWritingSessionRepository | None:
+    if session_factory is None:
+        return None
+    return AIWritingSessionRepository(session_factory)
+
+
+__all__ = ["AIWritingSessionRow", "AIWritingSessionRepository", "make_ai_writing_session_store"]
@@ -0,0 +1,26 @@
+"""ORM model for AI writing sessions."""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+
+from sqlalchemy import DateTime, String, Text
+from sqlalchemy.orm import Mapped, mapped_column
+
+from deerflow.persistence.base import Base
+
+
+class AIWritingSessionRow(Base):
+    __tablename__ = "ai_writing_sessions"
+
+    id: Mapped[str] = mapped_column(String(64), primary_key=True)
+    user_id: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)
+    title: Mapped[str] = mapped_column(String(512), nullable=False, default="")
+    user_intent: Mapped[str | None] = mapped_column(String(512), nullable=True)
+    status: Mapped[str] = mapped_column(String(32), nullable=False, default="in_progress")
+    draft_title: Mapped[str | None] = mapped_column(String(512), nullable=True)
+    draft_markdown: Mapped[str | None] = mapped_column(Text, nullable=True)
+    completed_interventions: Mapped[str | None] = mapped_column(Text, nullable=True)  # JSON array
+    review_result: Mapped[str | None] = mapped_column(Text, 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))
@@ -0,0 +1,154 @@
+"""SQLAlchemy-backed AI writing session storage."""
+
+from __future__ import annotations
+
+import json
+from datetime import UTC, datetime
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+from deerflow.persistence.ai_writing_sessions.model import AIWritingSessionRow
+from deerflow.runtime.user_context import AUTO, _AutoSentinel, resolve_user_id
+
+_UNSET = object()
+
+
+class AIWritingSessionRepository:
+    def __init__(self, session_factory: async_sessionmaker[AsyncSession]) -> None:
+        self._sf = session_factory
+
+    @staticmethod
+    def _row_to_dict(row: AIWritingSessionRow) -> dict:
+        d = row.to_dict()
+        for k in ("created_at", "updated_at"):
+            if isinstance(d.get(k), datetime):
+                d[k] = d[k].isoformat()
+        for k in ("completed_interventions", "review_result"):
+            val = d.get(k)
+            if isinstance(val, str):
+                try:
+                    d[k] = json.loads(val)
+                except Exception:
+                    d[k] = None
+        return d
+
+    async def create(
+        self,
+        *,
+        id: str,
+        title: str,
+        user_intent: str | None = None,
+        user_id: str | None | _AutoSentinel = AUTO,
+    ) -> dict:
+        resolved_user_id = resolve_user_id(user_id, method_name="AIWritingSessionRepository.create")
+        now = datetime.now(UTC)
+        row = AIWritingSessionRow(
+            id=id,
+            user_id=resolved_user_id,
+            title=title,
+            user_intent=user_intent,
+            status="in_progress",
+            draft_title=None,
+            draft_markdown=None,
+            completed_interventions=None,
+            review_result=None,
+            created_at=now,
+            updated_at=now,
+        )
+        async with self._sf() as session:
+            session.add(row)
+            await session.commit()
+            await session.refresh(row)
+            return self._row_to_dict(row)
+
+    async def update(
+        self,
+        session_id: str,
+        *,
+        user_id: str | None | _AutoSentinel = AUTO,
+        title=_UNSET,
+        status=_UNSET,
+        draft_title=_UNSET,
+        draft_markdown=_UNSET,
+        completed_interventions=_UNSET,
+        review_result=_UNSET,
+    ) -> dict | None:
+        resolved_user_id = resolve_user_id(user_id, method_name="AIWritingSessionRepository.update")
+        async with self._sf() as session:
+            row = await session.get(AIWritingSessionRow, session_id)
+            if row is None:
+                return None
+            if resolved_user_id is not None and row.user_id != resolved_user_id:
+                return None
+            if title is not _UNSET:
+                row.title = title
+            if status is not _UNSET:
+                row.status = status
+            if draft_title is not _UNSET:
+                row.draft_title = draft_title
+            if draft_markdown is not _UNSET:
+                row.draft_markdown = draft_markdown
+            if completed_interventions is not _UNSET:
+                row.completed_interventions = (
+                    json.dumps(completed_interventions, ensure_ascii=False)
+                    if completed_interventions is not None
+                    else None
+                )
+            if review_result is not _UNSET:
+                row.review_result = (
+                    json.dumps(review_result, ensure_ascii=False)
+                    if review_result is not None
+                    else None
+                )
+            row.updated_at = datetime.now(UTC)
+            await session.commit()
+            await session.refresh(row)
+            return self._row_to_dict(row)
+
+    async def get(
+        self,
+        session_id: str,
+        *,
+        user_id: str | None | _AutoSentinel = AUTO,
+    ) -> dict | None:
+        resolved_user_id = resolve_user_id(user_id, method_name="AIWritingSessionRepository.get")
+        async with self._sf() as session:
+            row = await session.get(AIWritingSessionRow, session_id)
+            if row is None:
+                return None
+            if resolved_user_id is not None and row.user_id != resolved_user_id:
+                return None
+            return self._row_to_dict(row)
+
+    async def list(
+        self,
+        *,
+        user_id: str | None | _AutoSentinel = AUTO,
+        limit: int = 50,
+    ) -> list[dict]:
+        resolved_user_id = resolve_user_id(user_id, method_name="AIWritingSessionRepository.list")
+        stmt = select(AIWritingSessionRow)
+        if resolved_user_id is not None:
+            stmt = stmt.where(AIWritingSessionRow.user_id == resolved_user_id)
+        stmt = stmt.order_by(AIWritingSessionRow.updated_at.desc()).limit(limit)
+        async with self._sf() as session:
+            result = await session.execute(stmt)
+            return [self._row_to_dict(r) for r in result.scalars()]
+
+    async def delete(
+        self,
+        session_id: str,
+        *,
+        user_id: str | None | _AutoSentinel = AUTO,
+    ) -> bool:
+        resolved_user_id = resolve_user_id(user_id, method_name="AIWritingSessionRepository.delete")
+        async with self._sf() as session:
+            row = await session.get(AIWritingSessionRow, session_id)
+            if row is None:
+                return False
+            if resolved_user_id is not None and row.user_id != resolved_user_id:
+                return False
+            await session.delete(row)
+            await session.commit()
+            return True
@@ -0,0 +1,50 @@
+"""Create ai_writing_sessions table.
+
+Revision ID: 20260517_01
+Revises: 20260515_01
+Create Date: 2026-05-17
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy import inspect
+
+
+revision = "20260517_01"
+down_revision = "20260515_01"
+branch_labels = None
+depends_on = None
+
+
+def _has_table(table_name: str) -> bool:
+    inspector = inspect(op.get_bind())
+    return table_name in inspector.get_table_names()
+
+
+def upgrade() -> None:
+    if _has_table("ai_writing_sessions"):
+        return
+
+    op.create_table(
+        "ai_writing_sessions",
+        sa.Column("id", sa.String(64), primary_key=True),
+        sa.Column("user_id", sa.String(64), nullable=True),
+        sa.Column("title", sa.String(512), nullable=False, server_default=""),
+        sa.Column("user_intent", sa.String(512), nullable=True),
+        sa.Column("status", sa.String(32), nullable=False, server_default="in_progress"),
+        sa.Column("draft_title", sa.String(512), nullable=True),
+        sa.Column("draft_markdown", sa.Text, nullable=True),
+        sa.Column("completed_interventions", sa.Text, nullable=True),
+        sa.Column("review_result", sa.Text, nullable=True),
+        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
+    )
+    op.create_index("ix_ai_writing_sessions_user_id", "ai_writing_sessions", ["user_id"])
+    op.create_index("ix_ai_writing_sessions_updated_at", "ai_writing_sessions", ["updated_at"])
+
+
+def downgrade() -> None:
+    if _has_table("ai_writing_sessions"):
+        op.drop_table("ai_writing_sessions")