feat：统一节点进度事件

app/processor/base.py

@@ -10,6 +10,7 @@ from pydantic import ValidationError
 if TYPE_CHECKING:
     from app.config import Settings
 from app.exceptions import AppError, ImportProcessError
+from app.processor.progress import node_progress
 
 T = TypeVar("T")
 
@@ -37,6 +38,7 @@ class BaseNode(ABC):
         start_time = time.perf_counter()
         self.logger.info("节点开始")
         try:
+            with node_progress(self.name, state):
                 result = self.process(state)
         except GraphInterrupt:
             raise
@@ -61,6 +63,7 @@ class BaseNode(ABC):
         start_time = time.perf_counter()
         self.logger.info("节点开始")
         try:
+            with node_progress(self.name, state):
                 result = await self.process(state)
         except GraphInterrupt:
             raise

app/processor/progress.py

@@ -0,0 +1,75 @@
+"""通过 LangGraph custom 流发送进度，不额外保存任务状态"""
+
+import asyncio
+from collections.abc import Mapping
+from contextlib import contextmanager
+from time import perf_counter
+from uuid import uuid4
+
+from langgraph.config import get_config, get_stream_writer
+from langgraph.errors import GraphInterrupt
+
+from app.schemas.node_progress import NodeProgress, NodeProgressStatus
+
+NODE_PROGRESS_MESSAGES = {
+    "requirement_node": "正在理解你的需求",
+    "qa_node": "正在整理回答",
+    "component_resource_node": "正在选择合适的图表和组件",
+    "planner_node": "正在设计布局",
+    "action_node": "正在准备画布修改",
+    "validator_node": "正在检查修改结果",
+    "reflection_node": "正在调整修改方案",
+}
+
+
+@contextmanager
+def node_progress(node: str, state):
+    """每次调用独立计时和编号，同步与异步入口共用"""
+    try:
+        config = get_config()
+    except RuntimeError:
+        # 独立调用节点没有图上下文，直接执行业务逻辑
+        yield
+        return
+
+    writer = get_stream_writer()
+    values = state if isinstance(state, Mapping) else {}
+    # 以实际检查点配置为准，兼容没有在 state 中填写 ID 的本地示例
+    thread_id = config.get("configurable", {}).get("thread_id") or values.get(
+        "thread_id"
+    )
+    execution_id = str(uuid4())
+    started = perf_counter()
+
+    def emit(status: NodeProgressStatus):
+        event = NodeProgress(
+            thread_id=thread_id,
+            message_id=values.get("message_id"),
+            execution_id=execution_id,
+            node=node,
+            label=NODE_PROGRESS_MESSAGES.get(node),
+            status=status,
+            duration_ms=None
+            if status == "started"
+            else round((perf_counter() - started) * 1000, 3),
+        )
+        writer(event.model_dump(mode="json", exclude_none=True))
+
+    emit("started")
+    try:
+        yield
+    except GraphInterrupt:
+        # interrupt 是正常暂停，恢复后会产生新的执行 ID
+        emit("waiting_for_input")
+        raise
+    except asyncio.CancelledError:
+        # 协程被取消也要结束这次节点执行，继续向上传递取消信号
+        emit("cancelled")
+        raise
+    except Exception:
+        # 只通知失败，异常详情仍由原有日志和 API 异常处理负责
+        emit("failed")
+        raise
+    else:
+        # 校验不通过等业务结果仍属于节点正常执行完成
+        emit("completed")

app/processor/state.py

@@ -16,6 +16,9 @@ from app.schemas.screen_plan import ScreenPlan
 
 
 class QueryState(TypedDict):
+    conversation_id: NotRequired[str]  # 当前聊天会话 ID
+    thread_id: NotRequired[str]  # 当前任务 ID
+    message_id: NotRequired[str]  # 本次触发执行的用户消息 ID，澄清恢复时更新
     request: JsonObject  # 原始请求，保留提交时的内容
     input_context: NotRequired[InputContext]  # 校验后的上下文，包含最新大屏和澄清历史
     intent_decision: NotRequired[DecisionResponse | None]  # Jev 原始意图判断及分数

app/schemas/agent.py

@@ -16,7 +16,9 @@ class ResumeRequest(BaseModel):
 
 
 class AgentResult(BaseModel):
-    thread_id: str
+    conversation_id: str  # 当前聊天会话 ID
+    thread_id: str  # 当前任务 ID
+    message_id: str  # 本次用户输入的 ID，不是助手回复 ID
     status: Literal["completed", "waiting_for_input", "failed", "cancelled"]
     actions: list[Action] = Field(default_factory=list)
     answer: str | None = None

app/schemas/node_progress.py

@@ -0,0 +1,20 @@
+"""节点进度事件"""
+
+from typing import Literal
+
+from pydantic import BaseModel, Field
+
+NodeProgressStatus = Literal[
+    "started", "completed", "waiting_for_input", "failed", "cancelled"
+]
+
+
+class NodeProgress(BaseModel):
+    type: Literal["node_progress"] = "node_progress"  # 用于区分流中的事件类型
+    thread_id: str | None = None  # 任务 ID，独立运行节点时可以为空
+    message_id: str | None = None  # 当前用户输入 ID
+    execution_id: str  # 单次节点执行 ID，开始和结束共用
+    node: str  # 稳定的技术标识，供程序处理和排查
+    label: str | None = None  # 用户进度提示，缺省时不展示该节点
+    status: NodeProgressStatus  # 节点执行状态，不代表任务总体状态
+    duration_ms: float | None = Field(default=None, ge=0)  # 结束时的执行耗时

app/services/agent.py

@@ -19,10 +19,20 @@ class AgentService:
     async def run(self, request: DesignRequest) -> AgentResult:
         # 新请求始终独立建任务，不用 conversation_id 充当检查点 ID
         thread_id = str(uuid4())
+        # 暂无会话创建接口，未传会话 ID 时先分配标识，不创建数据库记录
+        conversation_id = request.context.conversation_id or str(uuid4())
+        payload = request.model_dump(mode="json")
+        payload["context"]["conversation_id"] = conversation_id
+
         return await self._invoke(
             thread_id,
-            {"request": request.model_dump(mode="json")},
+            {
-            request.context.conversation_id,
+                "request": payload,
+                "conversation_id": conversation_id,
+                "thread_id": thread_id,
+                "message_id": str(uuid4()),
+            },
+            conversation_id,
         )
 
     async def resume(self, request: ResumeRequest) -> AgentResult:
@@ -46,11 +56,28 @@ class AgentService:
                 and reply.option_id not in {item.id for item in clarification.options}
             ):
                 raise BizException(422, "请选择当前问题的有效选项")
-            session_id = (
+
-                snapshot.values["request"].get("context", {}).get("conversation_id")
+            conversation_id = snapshot.values["conversation_id"]
+            # 恢复同一任务不能切换会话，补充编辑上下文时也保留原会话标识
+            if reply.context is not None:
+                if reply.context.conversation_id not in (None, conversation_id):
+                    raise BizException(422, "恢复任务时不能切换会话")
+                reply = reply.model_copy(
+                    update={
+                        "context": reply.context.model_copy(
+                            update={"conversation_id": conversation_id}
                         )
+                    }
+                )
+
+            # 每次有效补充或取消生成新的输入 ID，任务和会话保持不变
             return await self._execute(
-                thread_id, Command(resume=reply.model_dump(mode="json")), session_id
+                thread_id,
+                Command(
+                    resume=reply.model_dump(mode="json"),
+                    update={"message_id": str(uuid4())},
+                ),
+                conversation_id,
             )
         finally:
             self._active.discard(thread_id)
@@ -91,7 +118,9 @@ class AgentService:
 
         # 只交付校验通过的完整动作批次，失败草稿留在后端
         return AgentResult(
+            conversation_id=state["conversation_id"],
             thread_id=thread_id,
+            message_id=state["message_id"],
             status="waiting_for_input" if waiting else state["status"],
             actions=state.get("actions") or []
             if state.get("validation_passed") is True and state["status"] == "completed"