From 62a5b2a8f4855fc3a1c1162c406717eb62ccc594 Mon Sep 17 00:00:00 2001 From: Marcus <1922576605@qq.com> Date: Fri, 2 Oct 2026 08:47:24 -0700 Subject: [PATCH] =?UTF-8?q?feat=EF=BC=9A=E5=AE=8C=E6=88=90=E9=87=8D?= =?UTF-8?q?=E6=9E=84=E5=92=8C=E6=B5=8B=E8=AF=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .env.example | 4 +++ app/actions/batch.py | 12 ++++++-- app/agent/screen_agent.py | 35 +++++++++++++++------ app/agent/tools.py | 8 ++++- app/config.py | 10 ++++-- app/processor/history.py | 39 ++++++++++++++++++++++++ app/processor/nodes/intent_node.py | 4 ++- app/processor/nodes/qa_node.py | 8 ++++- app/processor/nodes/screen_agent_node.py | 5 ++- app/processor/summary.py | 8 +++-- app/prompt/qa.py | 4 +-- app/schemas/actions/add.py | 3 +- app/schemas/conversation.py | 2 +- app/schemas/input_context.py | 2 +- app/services/agent.py | 37 +++++++++++----------- 15 files changed, 138 insertions(+), 43 deletions(-) create mode 100644 app/processor/history.py diff --git a/.env.example b/.env.example index 492a946..0940792 100644 --- a/.env.example +++ b/.env.example @@ -16,6 +16,10 @@ INTENT_QA_MIN_PROBABILITY=0.5 # screen agent 单次运行的模型调用上限，包括读取 skill、查询物料、提交修正和最终回复 MAX_AGENT_STEPS=12 +# 会话历史：会话记录完整保存，每次请求最多读取最近 20 轮，再按 token 预算从最近一轮往前保留完整的轮次 +CONVERSATION_HISTORY_TURNS=20 +HISTORY_MAX_TOKENS=4096 + # MongoDB 会话记忆与检查点持久化，不配置则使用进程内存，重启后丢失 # MONGODB_URI=mongodb://127.0.0.1:27017 MONGODB_DB=screen_agent diff --git a/app/actions/batch.py b/app/actions/batch.py index 1d3781f..048b580 100644 --- a/app/actions/batch.py +++ b/app/actions/batch.py @@ -39,13 +39,21 @@ def merge_config(defaults: dict, overrides: dict) -> dict: def complete_add(action: AddAction, catalog: ComponentCatalog) -> AddAction: - """按 node.type 找物料补全新增组件，模型显式写出的字段优先，并按物料设置 sample_data""" + """按 node.type 找物料补全新增组件，模型显式写出的字段优先 + + sample_data 模型写了就以模型为准（只有它知道是否换成了自己的数据），没写时按物料设置 + """ material = find_material(catalog, action.node.type) if not DATA_BINDING_SUPPORTED and action.node.model_extra.get("dataId"): raise ValueError(DATA_BINDING_ERROR) overrides = action.node.model_dump(mode="json", exclude_unset=True) node = ScreenNode.model_validate(merge_config(material.template(), overrides)) - return action.model_copy(update={"node": node, "sample_data": material.sample_data}) + sample_data = ( + action.sample_data + if "sample_data" in action.model_fields_set + else material.sample_data + ) + return action.model_copy(update={"node": node, "sample_data": sample_data}) def run_batch( diff --git a/app/agent/screen_agent.py b/app/agent/screen_agent.py index 4b4fc95..8329e82 100644 --- a/app/agent/screen_agent.py +++ b/app/agent/screen_agent.py @@ -10,7 +10,7 @@ from typing import Literal, Self from langchain.agents import create_agent from langchain.agents.middleware import ModelCallLimitMiddleware, wrap_model_call from langchain.agents.structured_output import ToolStrategy -from langchain_core.messages import AIMessage +from langchain_core.messages import AIMessage, HumanMessage from loguru import logger from pydantic import BaseModel, Field, model_validator @@ -62,21 +62,36 @@ class AgentReply(BaseModel): return self -# 偶发的空回复最多重新调用几次 -EMPTY_REPLY_RETRIES = 2 +# 模型没有调用工具时最多纠正几次 +NO_TOOL_CALL_RETRIES = 2 +NO_TOOL_CALL_HINT = ( + "每一轮都要调用工具。要结束时调用 AgentReply，给用户的话写在 message 里" +) @wrap_model_call -async def retry_empty_reply(request, handler): - """每一轮都必须调用工具（最后一轮是 AgentReply）；模型偶尔返回既没有文本也没有工具调用的空回复， - 这是调用失败而不是模型的决定，重新调用即可""" +async def require_tool_call(request, handler): + """ToolStrategy 要求每一轮都调用工具，结束时调用 AgentReply；模型偶尔只回文本（想直接回复用户）或空回复 + + 原样重试往往得到同样的结果，或者模型转而重复提交 apply_actions 直到调用次数用尽， + 所以把它的回复和纠正说明加在下一次调用里；只用于这一次调用，不写进 agent 的消息 + """ response = await handler(request) - for _ in range(EMPTY_REPLY_RETRIES): + for _ in range(NO_TOOL_CALL_RETRIES): message = response.result[0] if response.result else None if not isinstance(message, AIMessage) or message.tool_calls: break logger.warning( - f"模型返回了没有工具调用的回复，重新调用: {message.response_metadata}" + f"模型没有调用工具，纠正后重新调用: {str(message.content)[:200]}" + ) + # 空回复不放回去，部分接口不接受内容为空的助手消息 + replied = [message] if message.content else [] + request = request.override( + messages=[ + *request.messages, + *replied, + HumanMessage(content=NO_TOOL_CALL_HINT), + ] ) response = await handler(request) return response @@ -93,12 +108,13 @@ async def get_screen_agent(): model=get_llm(), tools=[get_nodes, apply_actions, *await get_mcp_tools()], system_prompt=SYSTEM_PROMPT, + # 用工具调用给出结论：试过原生结构化输出（ProviderStrategy），歧义时模型更常直接执行而不追问，评测多删了组件 response_format=ToolStrategy(AgentReply), state_schema=ScreenAgentState, context_schema=ScreenAgentContext, middleware=[ *skill_middleware(), - retry_empty_reply, + require_tool_call, # 超过上限时插入说明并结束，没有结构化结果，由节点按失败处理 ModelCallLimitMiddleware( run_limit=settings.max_agent_steps, exit_behavior="end" @@ -108,4 +124,5 @@ async def get_screen_agent(): checkpointer=False, name="screen_agent", ) + return _agent diff --git a/app/agent/tools.py b/app/agent/tools.py index 1ef4c57..c402980 100644 --- a/app/agent/tools.py +++ b/app/agent/tools.py @@ -10,6 +10,7 @@ from langchain.agents import AgentState from langchain.tools import ToolRuntime, tool from langchain_core.messages import ToolMessage from langgraph.types import Command +from pydantic import Field from app.actions.batch import BatchError, run_batch from app.schemas.actions import Action, AddAction @@ -63,7 +64,12 @@ def _result_text(actions: list[Action]) -> str: @tool def apply_actions( - actions: list[Action], runtime: ToolRuntime[ScreenAgentContext] + actions: Annotated[ + list[Action], + # 空的一批会清掉之前成功的结果；参数校验失败不改状态，之前的结果保留 + Field(min_length=1, description="至少一个动作；不需要修改时不要调用，直接回复"), + ], + runtime: ToolRuntime[ScreenAgentContext], ) -> Command: """提交本次要对大屏做的全部修改 diff --git a/app/config.py b/app/config.py index dbf2e49..0cd732c 100644 --- a/app/config.py +++ b/app/config.py @@ -58,9 +58,13 @@ class Settings: conversation_empty_ttl: int = field( default_factory=lambda: int(os.getenv("CONVERSATION_EMPTY_TTL", "7200")) ) - # 每次请求带入模型的最近对话轮数 + # 每次请求从会话记录读取的最近轮数上限，实际带入多少由 history_max_tokens 决定 conversation_history_turns: int = field( - default_factory=lambda: int(os.getenv("CONVERSATION_HISTORY_TURNS", "6")) + default_factory=lambda: int(os.getenv("CONVERSATION_HISTORY_TURNS", "20")) + ) + # 每次请求带入的会话历史 token 预算（近似计数），从最近一轮往前保留完整的轮次 + history_max_tokens: int = field( + default_factory=lambda: int(os.getenv("HISTORY_MAX_TOKENS", "4096")) ) # MongoDB 连接串，未配置时会话记录和检查点回退为进程内存实现 @@ -107,6 +111,8 @@ class Settings: raise ValueError("CONVERSATION_EMPTY_TTL 必须是正整数") if self.conversation_history_turns < 0: raise ValueError("CONVERSATION_HISTORY_TURNS 必须是非负整数") + if self.history_max_tokens < 1: + raise ValueError("HISTORY_MAX_TOKENS 必须是正整数") if self.max_agent_steps < 1: raise ValueError("MAX_AGENT_STEPS 必须是正整数") if self.mcp_tool_timeout <= 0: diff --git a/app/processor/history.py b/app/processor/history.py new file mode 100644 index 0000000..c409547 --- /dev/null +++ b/app/processor/history.py @@ -0,0 +1,39 @@ +"""会话历史：数据库保存完整对话，每次请求按 token 预算只带最近的完整轮次 + +裁剪在服务层组装请求时做一次，意图判断、问答和 screen agent 看到的是同一段历史 +""" + +from langchain_core.messages import ( + AIMessage, + BaseMessage, + HumanMessage, + trim_messages, +) +from langchain_core.messages.utils import count_tokens_approximately + +from app.schemas.base import JsonObject + + +def history_messages(history: list[JsonObject]) -> list[BaseMessage]: + """每轮转成一条用户消息和一条助手消息，按时间从早到晚排列""" + messages: list[BaseMessage] = [] + for turn in history: + messages.append(HumanMessage(content=turn["user"])) + messages.append(AIMessage(content=turn["assistant"])) + return messages + + +def trim_history(history: list[JsonObject], max_tokens: int) -> list[JsonObject]: + """从最近一轮往前保留，总量不超过 max_tokens；只保留完整的轮次，不从一段回复中间截断 + + 单独一轮就超过预算时这一轮也不保留 + """ + kept = trim_messages( + history_messages(history), + max_tokens=max_tokens, + token_counter=count_tokens_approximately, + strategy="last", + # 从用户消息开始，助手回复放不下时连同它的用户消息一起去掉 + start_on="human", + ) + return history[len(history) - len(kept) // 2 :] diff --git a/app/processor/nodes/intent_node.py b/app/processor/nodes/intent_node.py index 84384d3..ddc3042 100644 --- a/app/processor/nodes/intent_node.py +++ b/app/processor/nodes/intent_node.py @@ -14,8 +14,10 @@ class IntentNode(BaseNode): async def process(self, state: QueryState) -> QueryState: updated_state: QueryState = state.copy() + context = state["input_context"] request = DecisionRequest( - state=request_summary(state["input_context"]), + # Jev 只接收 JSON，会话历史作为 history 字段和本次请求放在一起 + state={"history": context.history, **request_summary(context)}, decisions={"intent": build_intent_question()}, ) response = await get_decision_engine().run(request) diff --git a/app/processor/nodes/qa_node.py b/app/processor/nodes/qa_node.py index 05e1e6e..85ff964 100644 --- a/app/processor/nodes/qa_node.py +++ b/app/processor/nodes/qa_node.py @@ -6,6 +6,7 @@ from langgraph.config import get_stream_writer from app.clients.llm import get_llm from app.processor.base import BaseNode +from app.processor.history import history_messages from app.processor.state import QueryState from app.prompt.qa import QA_SYSTEM_PROMPT @@ -24,11 +25,16 @@ class QANode(BaseNode): parser = StrOutputParser() parts: list[str] = [] + context = state["input_context"] async for chunk in get_llm().astream( [ SystemMessage(content=QA_SYSTEM_PROMPT), + # 会话历史作为此前的对话消息，本次请求是最后一条用户消息 + *history_messages(context.history), HumanMessage( - content=state["input_context"].model_dump_json(exclude_unset=True) + content=context.model_dump_json( + exclude_unset=True, exclude={"history"} + ) ), ] ): diff --git a/app/processor/nodes/screen_agent_node.py b/app/processor/nodes/screen_agent_node.py index d150479..2536e50 100644 --- a/app/processor/nodes/screen_agent_node.py +++ b/app/processor/nodes/screen_agent_node.py @@ -10,6 +10,7 @@ from langchain_core.messages import HumanMessage from app.agent.screen_agent import AgentReply, get_screen_agent from app.agent.tools import ScreenAgentContext from app.processor.base import BaseNode +from app.processor.history import history_messages from app.processor.state import QueryState from app.processor.summary import request_summary from app.resources.component import get_catalog_cache @@ -55,10 +56,12 @@ class ScreenAgentNode(BaseNode): agent = await get_screen_agent() result = await agent.ainvoke( { + # 会话历史作为此前的对话消息，本次请求是最后一条用户消息 "messages": [ + *history_messages(context.history), HumanMessage( content=json.dumps(request_summary(context), ensure_ascii=False) - ) + ), ] }, context=ScreenAgentContext(screen=context.screen, catalog=catalog), diff --git a/app/processor/summary.py b/app/processor/summary.py index 091fc64..5cb5079 100644 --- a/app/processor/summary.py +++ b/app/processor/summary.py @@ -1,11 +1,14 @@ -"""请求摘要：意图判断和 screen agent 共用，只给组件的识别信息和布局，完整配置按需读取""" +"""请求摘要：意图判断和 screen agent 共用，只给组件的识别信息和布局，完整配置按需读取 + +只描述本次请求；会话历史由各使用方另行传入，见 history.py +""" from app.schemas.base import JsonObject from app.schemas.input_context import InputContext def request_summary(context: InputContext) -> JsonObject: - """用户输入、会话历史、澄清记录、编辑上下文、附件和大屏摘要""" + """用户输入、澄清记录、编辑上下文、附件和大屏摘要""" nodes: list[JsonObject] = [] for node in context.screen.nodes: # 图表标题帮助判断用户指的是哪个组件；没有 option 或标题格式不符时按无标题处理 @@ -28,7 +31,6 @@ def request_summary(context: InputContext) -> JsonObject: 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/prompt/qa.py b/app/prompt/qa.py index 49aa637..7418a7a 100644 --- a/app/prompt/qa.py +++ b/app/prompt/qa.py @@ -11,8 +11,8 @@ QA_SYSTEM_PROMPT = f"""## 任务 可以简短提示还能继续帮忙，不列举能力清单 ## 回答依据 -结合当前问题、大屏快照、history（此前各轮的用户输入与助手摘要）和 clarification_history，以用户后续明确补充或纠正为准 -理解“刚才那个”“再大一点”等指代时参考 history，但当前大屏快照才是最新事实 +结合当前问题、大屏快照、此前的对话和 clarification_history，以用户后续明确补充或纠正为准 +理解“刚才那个”“再大一点”等指代时参考此前的对话，但当前大屏快照才是最新事实 仅在与请求相关时使用大屏快照；一般交流无需强行关联大屏或要求用户补充大屏信息 区分快照事实与待验证的推测，证据不足时说明缺少的信息，不编造故障原因 diff --git a/app/schemas/actions/add.py b/app/schemas/actions/add.py index a4b06b5..bcff2fd 100644 --- a/app/schemas/actions/add.py +++ b/app/schemas/actions/add.py @@ -19,5 +19,6 @@ class AddAction(ActionModel): ) sample_data: bool = Field( default=False, - description="由后端按物料设置，不用填写；true 表示组件使用物料示例数据，不是真实业务数据", + description="true 表示组件显示的不是真实业务数据（物料示例数据，或自己编写的数据）。不写时按物料设置；" + "只有展示的数据全部来自大屏上已有的组件时写 false，例如更换类型时迁移了原组件的数据", ) diff --git a/app/schemas/conversation.py b/app/schemas/conversation.py index d0ae87c..665f861 100644 --- a/app/schemas/conversation.py +++ b/app/schemas/conversation.py @@ -14,7 +14,7 @@ class Turn(BaseModel): message_id: str prompt: str # 用户原始输入 outcome: Literal["completed", "waiting_for_input", "failed", "cancelled"] - summary: str # 助手侧摘要，只描述后端生成了什么，不代表前端已执行 + summary: str # 助手一侧的完整回复（回答、追问或失败说明），不截断；不代表前端已执行 created_at: datetime diff --git a/app/schemas/input_context.py b/app/schemas/input_context.py index 7ce1388..e329770 100644 --- a/app/schemas/input_context.py +++ b/app/schemas/input_context.py @@ -10,5 +10,5 @@ class InputContext(DesignRequest): """当前复用请求协议，后续可添加解析结果而不改变前端传参""" clarification_history: list[JsonObject] = Field(default_factory=list) - # 同一会话此前各轮的用户输入与助手摘要，由服务端按 conversation_id 写入，按时间从早到晚排列 + # 同一会话此前各轮的用户输入与助手回复，由服务端按 conversation_id 读取并按 token 预算裁剪，按时间从早到晚排列 history: list[JsonObject] = Field(default_factory=list) diff --git a/app/services/agent.py b/app/services/agent.py index e4e6e13..e7427cb 100644 --- a/app/services/agent.py +++ b/app/services/agent.py @@ -10,19 +10,13 @@ from loguru import logger from app.config import settings from app.core.exceptions import BizException from app.observability.tracing import astream_screen +from app.processor.history import trim_history 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, store: ConversationStore): self.graph = graph @@ -51,11 +45,18 @@ class AgentService: raise payload = request.model_dump(mode="json") - # 历史只信任服务端记录，覆盖前端可能携带的同名扩展字段 - payload["history"] = [ - {"user": turn.prompt, "assistant": turn.summary, "outcome": turn.outcome} - for turn in turns - ] + # 历史只信任服务端记录，覆盖前端可能携带的同名扩展字段；记录是完整的，这里按 token 预算裁剪 + payload["history"] = trim_history( + [ + { + "user": turn.prompt, + "assistant": turn.summary, + "outcome": turn.outcome, + } + for turn in turns + ], + settings.history_max_tokens, + ) return ( thread_id, @@ -144,17 +145,17 @@ class AgentService: self._conversation_of[thread_id] = conversation_id @staticmethod - def _summarize(state, result: AgentResult) -> str: - """用一句话描述本轮结果，只陈述后端生成了什么，不代表前端已执行""" + def _summarize(result: AgentResult) -> str: + """本轮助手一侧的完整文本，只陈述后端生成了什么，不代表前端已执行；不截断，带入请求时再按预算裁剪""" if result.status == "waiting_for_input": question = result.clarification.question if result.clarification else "" - return _clip(f"向用户提问：{question}") + return f"向用户提问：{question}" if result.status == "cancelled": return "用户取消了本次补充" if result.status == "failed": - return _clip(f"处理失败：{result.answer or '未知原因'}") + return f"处理失败：{result.answer or '未知原因'}" if result.answer: - return _clip(result.answer) + return result.answer return f"已生成 {len(result.actions)} 个画布动作" async def _save_turn(self, state, result: AgentResult) -> None: @@ -167,7 +168,7 @@ class AgentService: message_id=result.message_id, prompt=state["request"]["prompt"], outcome=result.status, - summary=self._summarize(state, result), + summary=self._summarize(result), created_at=datetime.now(timezone.utc), ) )