diff --git a/app/processor/base.py b/app/processor/base.py index 7706a8c..9cde8d3 100644 --- a/app/processor/base.py +++ b/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,7 +38,8 @@ class BaseNode(ABC): start_time = time.perf_counter() self.logger.info("节点开始") try: - result = self.process(state) + with node_progress(self.name, state): + result = self.process(state) except GraphInterrupt: raise except (AppError, ValidationError): @@ -61,7 +63,8 @@ class BaseNode(ABC): start_time = time.perf_counter() self.logger.info("节点开始") try: - result = await self.process(state) + with node_progress(self.name, state): + result = await self.process(state) except GraphInterrupt: raise except (AppError, ValidationError): diff --git a/app/processor/progress.py b/app/processor/progress.py new file mode 100644 index 0000000..b85b18d --- /dev/null +++ b/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") diff --git a/app/processor/state.py b/app/processor/state.py index 994ab4d..b5ea818 100644 --- a/app/processor/state.py +++ b/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 原始意图判断及分数 diff --git a/app/schemas/agent.py b/app/schemas/agent.py index 25fa856..03c4b5b 100644 --- a/app/schemas/agent.py +++ b/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 diff --git a/app/schemas/node_progress.py b/app/schemas/node_progress.py new file mode 100644 index 0000000..93951ef --- /dev/null +++ b/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) # 结束时的执行耗时 diff --git a/app/services/agent.py b/app/services/agent.py index ef895aa..222ae29 100644 --- a/app/services/agent.py +++ b/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"