From a89a458c225b04cc1cca9741a28a05d83fa473cd Mon Sep 17 00:00:00 2001 From: Marcus <1922576605@qq.com> Date: Tue, 29 Sep 2026 09:55:17 -0700 Subject: [PATCH] =?UTF-8?q?feat=EF=BC=9A=E4=BC=98=E5=8C=96=E6=B5=81?= =?UTF-8?q?=E5=BC=8F=E8=BE=93=E5=87=BA=EF=BC=8C=E5=A2=9E=E5=8A=A0=E4=B8=B4?= =?UTF-8?q?=E6=97=B6=E5=A4=9A=E8=BD=AE=E5=AF=B9=E8=AF=9D=E8=AE=B0=E5=BD=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/api/agent.py | 4 +- app/config.py | 13 ++++ app/main.py | 9 ++- app/processor/nodes/intent_node.py | 1 + app/processor/nodes/qa_node.py | 32 +++++++-- app/prompt/intent.py | 2 +- app/prompt/qa.py | 3 +- app/prompt/requirement.py | 8 ++- app/schemas/input_context.py | 2 + app/services/agent.py | 102 +++++++++++++++++++++++++---- 10 files changed, 153 insertions(+), 23 deletions(-) diff --git a/app/api/agent.py b/app/api/agent.py index 06a01d9..acbdc28 100644 --- a/app/api/agent.py +++ b/app/api/agent.py @@ -12,7 +12,9 @@ router = APIRouter(prefix="/agent", tags=["Agent"]) @router.post("/run", response_class=AgentEventResponse, status_code=200) async def stream_agent(body: DesignRequest, request: Request): service = request.app.state.agent_service - return AgentEventResponse(service, service.stream_run(body), request) + # 在发送流响应头之前完成会话校验并占用任务 + execution = await service.prepare_run(body) + return AgentEventResponse(service, execution, request) @router.post("/resume", response_class=AgentEventResponse, status_code=200) diff --git a/app/config.py b/app/config.py index 20b9769..06c50e6 100644 --- a/app/config.py +++ b/app/config.py @@ -50,7 +50,20 @@ class Settings: default_factory=lambda: int(os.getenv("MAX_ACTION_RETRIES", "2")) ) + # 没有任何对话的空会话保留时长（秒），超时后视为不存在 + conversation_empty_ttl: int = field( + default_factory=lambda: int(os.getenv("CONVERSATION_EMPTY_TTL", "7200")) + ) + # 每次请求带入模型的最近对话轮数 + conversation_history_turns: int = field( + default_factory=lambda: int(os.getenv("CONVERSATION_HISTORY_TURNS", "6")) + ) + def __post_init__(self): + if self.conversation_empty_ttl <= 0: + raise ValueError("CONVERSATION_EMPTY_TTL 必须是正整数") + if self.conversation_history_turns < 0: + raise ValueError("CONVERSATION_HISTORY_TURNS 必须是非负整数") if type(self.max_action_retries) is not int or self.max_action_retries < 0: raise ValueError("MAX_ACTION_RETRIES 必须是非负整数") if not 0 <= self.intent_min_confidence <= 1: diff --git a/app/main.py b/app/main.py index 18b2db2..45f0c6b 100644 --- a/app/main.py +++ b/app/main.py @@ -7,6 +7,7 @@ from fastapi import FastAPI from loguru import logger from app.api.agent import router +from app.api.conversations import router as conversation_router from app.core.exceptions import register_exception_handlers from app.core.logging import setup_logging @@ -19,7 +20,12 @@ async def lifespan(application: FastAPI): from app.processor.graph import build_graph from app.services.agent import AgentService - application.state.agent_service = AgentService(build_graph()) + from app.services.conversation import InMemoryConversationStore + + # TODO 接入 MongoDB 后替换为持久化实现，并同步替换图的 checkpointer + store = InMemoryConversationStore() + application.state.conversation_store = store + application.state.agent_service = AgentService(build_graph(), store) try: yield finally: @@ -38,6 +44,7 @@ def create_app() -> FastAPI: # 统一处理请求校验、HTTP 错误和未捕获异常 register_exception_handlers(application) application.include_router(router) + application.include_router(conversation_router) return application diff --git a/app/processor/nodes/intent_node.py b/app/processor/nodes/intent_node.py index 0556fd3..8a4a1a8 100644 --- a/app/processor/nodes/intent_node.py +++ b/app/processor/nodes/intent_node.py @@ -50,6 +50,7 @@ class IntentNode(BaseNode): # 汇总用户需求、澄清历史、交互上下文和大屏摘要，作为意图判断的输入 return { "prompt": context.prompt, + "history": context.history, "clarification_history": context.clarification_history, "context": context.context.model_dump(mode="json", exclude_unset=True), "screen_summary": { diff --git a/app/processor/nodes/qa_node.py b/app/processor/nodes/qa_node.py index 738c1d6..46bcd33 100644 --- a/app/processor/nodes/qa_node.py +++ b/app/processor/nodes/qa_node.py @@ -2,6 +2,7 @@ from langchain_core.messages import HumanMessage, SystemMessage from langchain_core.output_parsers import StrOutputParser +from langgraph.config import get_stream_writer from app.clients.llm import get_llm from app.processor.base import BaseNode @@ -18,8 +19,16 @@ class QANode(BaseNode): # 获取普通对话模型，问答只需文本回答，不绑定结构化输出协议 llm_client = get_llm() - # 异步调用模型，结合用户问题、当前大屏和澄清记录生成回答 - llm_res = await llm_client.ainvoke( + try: + # 图执行上下文中通过 custom 流逐段推送回答，独立调用节点时没有写入器 + writer = get_stream_writer() + except RuntimeError: + writer = None + + parser = StrOutputParser() + parts: list[str] = [] + # 异步流式调用模型，结合用户问题、当前大屏和澄清记录生成回答 + async for chunk in llm_client.astream( [ # 系统消息提供稳定的问答规则和回答边界 SystemMessage(content=QA_SYSTEM_PROMPT), @@ -28,10 +37,23 @@ class QANode(BaseNode): content=state["input_context"].model_dump_json(exclude_unset=True) ), ] - ) + ): + # 从每个片段提取文本，兼容文本内容块，工具调用等无文本片段会得到空串 + delta = parser.invoke(chunk) + if not delta: + continue + parts.append(delta) + if writer is not None: + writer( + { + "type": "answer_delta", + "message_id": state.get("message_id"), + "delta": delta, + } + ) - # 从模型响应中提取文本，兼容文本内容块，并去除回答首尾的空白 - answer = StrOutputParser().invoke(llm_res).strip() + # 拼接完整回答，并去除首尾的空白 + answer = "".join(parts).strip() # 如果没有有效文本，抛出异常，避免将空回答标记为处理完成 if not answer: raise ValueError("问答模型未返回文本回答") diff --git a/app/prompt/intent.py b/app/prompt/intent.py index 98f1803..26f67eb 100644 --- a/app/prompt/intent.py +++ b/app/prompt/intent.py @@ -9,7 +9,7 @@ def build_intent_question() -> ChoiceDecision: instructions=( "Classify the user's primary intent. " "Classify the latest user answer in `clarification_history`, or `prompt` when there is no history. " - "Use earlier turns and the selected option to resolve references; the latest answer overrides earlier conflicting intent. " + "Use earlier turns, `history` (previous turns of this conversation) and the selected option to resolve references; the latest answer overrides earlier conflicting intent. " "Once the requested action or response is clear, earlier ambiguity does not require another confirmation. " "Choose modify_screen only when the user explicitly requests an actual edit. " "Choose qa for clear requests for a textual answer or general conversation, including information, explanation, diagnosis, advice, and assistant identity or capabilities. " diff --git a/app/prompt/qa.py b/app/prompt/qa.py index 5d0af69..9298219 100644 --- a/app/prompt/qa.py +++ b/app/prompt/qa.py @@ -5,7 +5,8 @@ QA_SYSTEM_PROMPT = """## 任务 也可回答助手身份、能力等一般交流请求；介绍自己时以大屏助手的职责和已知能力为依据，不编造模型身份或未提供的能力 ## 回答依据 -结合当前问题、大屏快照和 clarification_history，以用户后续明确补充或纠正为准 +结合当前问题、大屏快照、history（此前各轮的用户输入与助手摘要）和 clarification_history，以用户后续明确补充或纠正为准 +理解“刚才那个”“再大一点”等指代时参考 history，但当前大屏快照才是最新事实 仅在与请求相关时使用大屏快照；一般交流无需强行关联大屏或要求用户补充大屏信息 区分快照事实与待验证的推测，证据不足时说明缺少的信息，不编造故障原因 diff --git a/app/prompt/requirement.py b/app/prompt/requirement.py index 53bacb6..fec1f22 100644 --- a/app/prompt/requirement.py +++ b/app/prompt/requirement.py @@ -49,10 +49,14 @@ REQUIREMENT_INPUT_TEMPLATE = """历史会话： def build_requirement_input( - context: InputContext, intent: Intent, *, history_text: str = "未提供" + context: InputContext, intent: Intent, *, history_text: str | None = None ) -> str: - """整理本次模型输入，历史会话由调用方加载后传入""" + """整理本次模型输入，历史会话默认取输入上下文中由服务端写入的记录""" data = context.model_dump(mode="json", exclude_unset=True) + # 会话历史 + history = data.pop("history", []) + if history_text is None: + history_text = json.dumps(history, ensure_ascii=False) if history else "未提供" content = REQUIREMENT_INPUT_TEMPLATE.format( history_text=history_text, clarification_text=json.dumps( diff --git a/app/schemas/input_context.py b/app/schemas/input_context.py index 72e7cd1..7ce1388 100644 --- a/app/schemas/input_context.py +++ b/app/schemas/input_context.py @@ -10,3 +10,5 @@ class InputContext(DesignRequest): """当前复用请求协议，后续可添加解析结果而不改变前端传参""" clarification_history: list[JsonObject] = Field(default_factory=list) + # 同一会话此前各轮的用户输入与助手摘要，由服务端按 conversation_id 写入，按时间从早到晚排列 + history: list[JsonObject] = Field(default_factory=list) diff --git a/app/services/agent.py b/app/services/agent.py index 1e0cb10..c559f79 100644 --- a/app/services/agent.py +++ b/app/services/agent.py @@ -1,29 +1,61 @@ """异步执行与恢复大屏任务，当前使用单进程内存检查点""" from contextlib import aclosing +from datetime import datetime, timezone from uuid import uuid4 from langgraph.types import Command from loguru import logger +from app.config import settings from app.core.exceptions import BizException from app.observability.tracing import astream_screen from app.schemas.agent import AgentResult, ResumeRequest +from app.schemas.conversation import Turn from app.schemas.request import DesignRequest +from app.services.conversation import ConversationStore + + +SUMMARY_MAX_CHARS = 500 + + +def _clip(text: str) -> str: + return text if len(text) <= SUMMARY_MAX_CHARS else text[:SUMMARY_MAX_CHARS] + "…" class AgentService: - def __init__(self, graph): + def __init__(self, graph, store: ConversationStore): self.graph = graph + self.store = store + # 同一时刻，一个任务和一个会话都只允许一个执行，避免历史顺序错乱 self._active: set[str] = set() + self._conversation_of: dict[str, str] = {} + + async def prepare_run(self, request: DesignRequest): + # 会话 ID 必须由后端签发，不再为缺失的 ID 兜底生成 + conversation_id = request.context.conversation_id + if conversation_id is None: + raise BizException(422, "请先创建会话并传入 conversation_id") + if not await self.store.exists(conversation_id): + raise BizException(404, "会话不存在或已过期") - def _prepare_run(self, request: DesignRequest): # 新请求始终独立建任务，不用 conversation_id 充当检查点 ID thread_id = str(uuid4()) - # 暂无会话创建接口，未传会话 ID 时先分配标识，不创建数据库记录 - conversation_id = request.context.conversation_id or str(uuid4()) + self._claim(thread_id, conversation_id) + try: + turns = await self.store.load_turns( + conversation_id, settings.conversation_history_turns + ) + except BaseException: + self.release(thread_id) + raise + payload = request.model_dump(mode="json") - payload["context"]["conversation_id"] = conversation_id + # 历史只信任服务端记录，覆盖前端可能携带的同名扩展字段 + payload["history"] = [ + {"user": turn.prompt, "assistant": turn.summary, "outcome": turn.outcome} + for turn in turns + ] return ( thread_id, @@ -36,11 +68,6 @@ class AgentService: conversation_id, ) - def stream_run(self, request: DesignRequest): - execution = self._prepare_run(request) - self._claim(execution[0]) - return execution - async def prepare_resume(self, request: ResumeRequest): # 在发送流响应头之前完成校验并占用任务 thread_id = str(request.thread_id) @@ -65,6 +92,7 @@ class AgentService: raise BizException(422, "请选择当前问题的有效选项") conversation_id = snapshot.values["conversation_id"] + self._claim_conversation(thread_id, conversation_id) # 恢复同一任务不能切换会话，补充编辑上下文时也保留原会话标识 if reply.context is not None: if reply.context.conversation_id not in (None, conversation_id): @@ -92,11 +120,56 @@ class AgentService: def release(self, thread_id): self._active.discard(thread_id) + self._conversation_of.pop(thread_id, None) - def _claim(self, thread_id: str) -> None: + def _claim(self, thread_id: str, conversation_id: str | None = None) -> None: if thread_id in self._active: raise BizException(409, "任务正在执行，请勿重复提交") self._active.add(thread_id) + if conversation_id is not None: + try: + self._claim_conversation(thread_id, conversation_id) + except BaseException: + self.release(thread_id) + raise + + def _claim_conversation(self, thread_id: str, conversation_id: str) -> None: + if conversation_id in self._conversation_of.values(): + raise BizException(409, "该会话有任务正在执行，请稍后再试") + self._conversation_of[thread_id] = conversation_id + + @staticmethod + def _summarize(state, result: AgentResult) -> str: + """用一句话描述本轮结果，只陈述后端生成了什么，不代表前端已执行""" + if result.status == "waiting_for_input": + question = result.clarification.question if result.clarification else "" + return _clip(f"向用户提问：{question}") + if result.status == "cancelled": + return "用户取消了本次补充" + if result.status == "failed": + return _clip(f"处理失败：{result.failure_reason or '未知原因'}") + if result.answer: + return _clip(result.answer) + requirement = state.get("requirement") + goal = requirement.goal if requirement is not None else "修改大屏" + return _clip(f"{goal}（已生成 {len(result.actions)} 个画布动作）") + + async def _save_turn(self, state, result: AgentResult) -> None: + # 历史写入失败不影响本次结果交付，只记录日志 + try: + await self.store.save_turn( + Turn( + conversation_id=result.conversation_id, + thread_id=result.thread_id, + message_id=result.message_id, + prompt=state["request"]["prompt"], + outcome=result.status, + summary=self._summarize(state, result), + created_at=datetime.now(timezone.utc), + ) + ) + except Exception: + logger.exception("保存对话轮次失败") @staticmethod def _result(state, thread_id): @@ -129,6 +202,11 @@ class AgentService: state = data elif data.get("type") == "node_progress" and data.get("label"): yield "progress", data + elif data.get("type") == "answer_delta": + # 问答文本逐段推送，完整回答仍在 final 事件中返回 + yield "answer_delta", data if state is None: raise RuntimeError("图未返回状态") - yield "final", self._result(state, thread_id).model_dump(mode="json") + result = self._result(state, thread_id) + await self._save_turn(state, result) + yield "final", result.model_dump(mode="json")