diff --git a/backend/app/api/conversations.py b/backend/app/api/conversations.py index a64e1be2..96b078da 100644 --- a/backend/app/api/conversations.py +++ b/backend/app/api/conversations.py @@ -70,6 +70,7 @@ class TurnPatch(BaseModel): run_id: str | None = None status: str | None = Field(default=None, pattern="^(running|done|error)$") error: str | None = None + review: dict[str, Any] | None = None class TurnOut(BaseModel): @@ -82,6 +83,8 @@ class TurnOut(BaseModel): run_id: str | None = None status: str error: str + #: 复核结论。None = 这一轮没复核过，和「复核过、没发现问题」不是一回事 + review: dict[str, Any] | None = None created_at: Any = None model_config = {"from_attributes": True} @@ -286,7 +289,9 @@ async def patch_turn( if not turn or turn.conversation_id != conversation_id: raise HTTPException(404, f"轮次 {turn_id} 不存在") - for field in ("question", "answer", "explanation", "graph", "run_id", "status", "error"): + for field in ( + "question", "answer", "explanation", "graph", "run_id", "status", "error", "review", + ): value = getattr(payload, field) if value is not None: setattr(turn, field, value) @@ -310,7 +315,12 @@ def _answer_of(turn: ConversationTurn) -> str: if turn.status == "error" and turn.error: return f"（这一轮失败了：{turn.error[:200]}）" if len(text) > HISTORY_ANSWER_CHARS: - return text[:HISTORY_ANSWER_CHARS] + "…（略）" + text = text[:HISTORY_ANSWER_CHARS] + "…（略）" + # 复核判过不可信的答案，下一轮必须知道。只给正文的话，模型会拿一个已知 + # 有问题的结论当事实继续往下推，错误就这么一轮轮传下去了 + note = (turn.review or {}).get("note", "") if isinstance(turn.review, dict) else "" + if note: + text = f"{text}\n（复核提示：{note[:200]}）".strip() return text diff --git a/backend/app/api/copilot.py b/backend/app/api/copilot.py index 85a2bd0d..8836c280 100644 --- a/backend/app/api/copilot.py +++ b/backend/app/api/copilot.py @@ -994,3 +994,69 @@ def _extract_json(text: str) -> dict[str, Any]: except json.JSONDecodeError: continue raise ValueError("模型输出里找不到 JSON") + + +# -------------------------------------------------------------------------- +# 跑完之后的复核 +# -------------------------------------------------------------------------- + + +class ReviewIn(BaseModel): + run_id: str = Field(min_length=1, description="要复核哪次运行") + question: str = Field(default="", description="用户原本问的是什么") + provider: str | None = None + model: str | None = None + #: 默认只在扫出异常信号时才叫模型。打开它就每次都复核——排查时有用， + #: 平时不开：干净的运行没什么可复核的，代价却是每次问数都多一次调用 + always: bool = False + + +@router.post("/review") +async def review_run( + payload: ReviewIn, session: AsyncSession = Depends(get_session) +) -> dict[str, Any]: + """接住一次编排的产出，判断它能不能就这么交给用户。 + + 「跑完了」和「答得对」是两回事。运行状态只说明流程走完了，而过程中检索 + 降级、工具报错、agent 步数用满这些事，原先一件都不会进到答案里——用户拿到 + 一个看起来很完整的结论，却无从知道它是在什么条件下得出的。 + + 规则层先扫一遍（engine/review.py:scan），没扫出东西就到此为止、直接返回， + 不花一次模型调用；扫出来了才叫模型重组答案并把异常说破。 + """ + from app.db.models import Run, RunEvent + from app.engine import review as rv + + run = await session.get(Run, payload.run_id) + if not run: + raise HTTPException(404, f"运行 {payload.run_id} 不存在") + + rows = list((await session.execute( + select(RunEvent).where(RunEvent.run_id == payload.run_id).order_by(RunEvent.seq) + )).scalars()) + + output = run.output if isinstance(run.output, dict) else None + signals = rv.scan(rows, output) + answer = rv.answer_text(output) + + if not signals and not payload.always: + return rv.ReviewResult(verdict="ok", signals=[]).as_dict() + + try: + spec = await copilot_model_spec(session, GenerateIn( + instruction="review", provider=payload.provider, model=payload.model, + ), max_tokens=4096) + model, _ = await get_chat_model(session, spec) + except (ProviderNotConfigured, HTTPException) as e: + # 没配模型不该让这一轮挂掉：规则层已经知道出了什么事，照样说得出话 + return rv.ReviewResult( + verdict="annotated", + note="；".join(s.detail for s in signals) or str(e), + retry=rv.should_retry(signals), + signals=signals, severity=rv.worst(signals), + ).as_dict() + + result = await rv.review( + model, question=payload.question, answer=answer, signals=signals + ) + return result.as_dict() diff --git a/backend/app/db/base.py b/backend/app/db/base.py index 81d1ab80..b8c1592c 100644 --- a/backend/app/db/base.py +++ b/backend/app/db/base.py @@ -84,6 +84,9 @@ _COLUMN_MIGRATIONS: list[tuple[str, str, str]] = [ ("chunks", "token_len", "INTEGER DEFAULT 0"), ("documents", "status", "TEXT DEFAULT 'ready'"), ("documents", "error", "TEXT DEFAULT ''"), + # 存量轮次没有复核结论。NULL 正好表示"这一轮没复核过"，和"复核过、没问题" + # （存 verdict=ok）是两回事，前端要分得开 + ("conversation_turns", "review", "JSON"), ] diff --git a/backend/app/db/models.py b/backend/app/db/models.py index e6616aa2..290738fe 100644 --- a/backend/app/db/models.py +++ b/backend/app/db/models.py @@ -423,5 +423,9 @@ class ConversationTurn(Base, TimestampMixin): # running | done | error status: Mapped[str] = mapped_column(String(20), default="running") error: Mapped[str] = mapped_column(Text, default="") + # 跑完之后的复核结论（engine/review.py）：判定档位、给用户的异常说明、 + # 命中的信号清单。落库是因为它是答案的一部分——刷新页面后只剩一个 + # 看起来很完整的答案、而"它哪里不可靠"没了，比不复核更糟 + review: Mapped[dict[str, Any] | None] = mapped_column(default=None) conversation: Mapped[Conversation] = relationship(back_populates="turns") diff --git a/backend/app/engine/compiler.py b/backend/app/engine/compiler.py index 9ee85fb3..76252a7f 100644 --- a/backend/app/engine/compiler.py +++ b/backend/app/engine/compiler.py @@ -131,6 +131,7 @@ def _wrap(node: GraphNode, run_ctx: RunContext) -> Callable[[GraphState], Awaita EventType.LOG, level="warn", message=f"第 {attempt + 1} 次失败（{_describe(e)}），准备重试", + code="node_retry", ) await asyncio.sleep(backoff * (2**attempt)) continue diff --git a/backend/app/engine/nodes/control.py b/backend/app/engine/nodes/control.py index 338b26dc..4b3f9f6c 100644 --- a/backend/app/engine/nodes/control.py +++ b/backend/app/engine/nodes/control.py @@ -156,7 +156,7 @@ async def run_loop(state: GraphState, ctx: NodeContext) -> dict[str, Any]: if hit_limit: ctx.emit( EventType.LOG, level="warn", - message=f"循环达到上限 {max_iter} 次，强制退出", + message=f"循环达到上限 {max_iter} 次，强制退出", code="loop_limit", ) decision = "body" if keep_going else "done" diff --git a/backend/app/engine/nodes/human.py b/backend/app/engine/nodes/human.py index 5f64cc64..65fcc64c 100644 --- a/backend/app/engine/nodes/human.py +++ b/backend/app/engine/nodes/human.py @@ -127,7 +127,8 @@ async def run_validate(state: GraphState, ctx: NodeContext) -> dict[str, Any]: path = "/".join(str(p) for p in e.absolute_path) or "(根)" errors = [f"{path}: {e.message}"] - ctx.emit(EventType.LOG, level="warn", message=f"第 {attempt + 1} 次校验失败：{errors[0]}") + ctx.emit(EventType.LOG, level="warn", message=f"第 {attempt + 1} 次校验失败：{errors[0]}", + code="validate_retry") attempt += 1 if attempt > max_retries or not repair: break diff --git a/backend/app/engine/nodes/knowledge.py b/backend/app/engine/nodes/knowledge.py index 82197fb2..993d0c8d 100644 --- a/backend/app/engine/nodes/knowledge.py +++ b/backend/app/engine/nodes/knowledge.py @@ -123,7 +123,7 @@ async def run_retrieve(state: GraphState, ctx: NodeContext) -> dict[str, Any]: hits = hits[:limit] for note in degraded: # 检索悄悄少了一半能力，不说的话用户只会觉得"最近搜得不准" - ctx.emit(EventType.LOG, level="warn", message=note) + ctx.emit(EventType.LOG, level="warn", message=note, code="retrieve_degraded") min_score = float(ctx.cfg("min_score", 0.0) or 0.0) hits = [h for h in hits if h["score"] >= min_score] diff --git a/backend/app/engine/nodes/llm.py b/backend/app/engine/nodes/llm.py index 3bf5fb83..eb05ad85 100644 --- a/backend/app/engine/nodes/llm.py +++ b/backend/app/engine/nodes/llm.py @@ -170,7 +170,8 @@ async def _invoke_streaming( response = await model.ainvoke(messages) except Exception as e: # noqa: BLE001 ctx.emit(EventType.LOG, level="warn", - message=f"流式失败，回退非流式：{explain_model_error(e, model_id)}") + message=f"流式失败，回退非流式：{explain_model_error(e, model_id)}", + code="stream_fallback") try: response = await model.ainvoke(messages) except Exception as retry_error: # noqa: BLE001 @@ -240,7 +241,8 @@ async def run_llm(state: GraphState, ctx: NodeContext) -> dict[str, Any]: if not text and reasoning: # 思考型模型把额度全花在推理上了，正文是空的 —— 说清楚而不是抛个空结果给下游 ctx.emit(EventType.LOG, level="warn", - message="模型只输出了思考内容，正文为空。把 max_tokens 调大，或把 thinking 设为 off。") + message="模型只输出了思考内容，正文为空。把 max_tokens 调大，或把 thinking 设为 off。", + code="empty_completion") output = {"text": text, "thinking": reasoning} usage = _usage_of(response, model_id) @@ -400,7 +402,7 @@ async def run_agent(state: GraphState, ctx: NodeContext) -> dict[str, Any]: continue if fix_note: ctx.emit(EventType.LOG, level="warn", - message=f"工具 {name}：{fix_note}") + message=f"工具 {name}：{fix_note}", code="tool_args_fixed") if _needs_approval(ctx, name): ctx.emit( @@ -494,7 +496,7 @@ async def run_agent(state: GraphState, ctx: NodeContext) -> dict[str, Any]: # 一步一工具时步数消耗得比并行快得多，这里不说清楚，用户只会看到 # 一个没头没尾的答案，而不知道是被步数掐断的 final_text = f"{final_text}\n\n（{hint}）".strip() if final_text else f"（{hint}）" - ctx.emit(EventType.LOG, level="warn", message=hint) + ctx.emit(EventType.LOG, level="warn", message=hint, code="step_limit") total_usage["total_tokens"] = total_usage["input_tokens"] + total_usage["output_tokens"] result = { diff --git a/backend/app/engine/nodes/multi.py b/backend/app/engine/nodes/multi.py index 5cf702d6..6a0b14c2 100644 --- a/backend/app/engine/nodes/multi.py +++ b/backend/app/engine/nodes/multi.py @@ -162,7 +162,7 @@ async def run_supervisor(state: GraphState, ctx: NodeContext) -> dict[str, Any]: ctx.emit(EventType.TOOL_START, tool=tname, args=targs, agent=name, call_id=cid) if fix_note: ctx.emit(EventType.LOG, level="warn", - message=f"工具 {tname}：{fix_note}") + message=f"工具 {tname}：{fix_note}", code="tool_args_fixed") if args_error: # 参数就不对，没必要真调一次。把"它接受什么"喂回去让它改 content = args_error diff --git a/backend/app/engine/nodes/tools.py b/backend/app/engine/nodes/tools.py index e0f062f9..f5270e6c 100644 --- a/backend/app/engine/nodes/tools.py +++ b/backend/app/engine/nodes/tools.py @@ -60,7 +60,7 @@ async def run_tool(state: GraphState, ctx: NodeContext) -> dict[str, Any]: # 替它跑通了这一次，但节点配置里那个错的参数名原封不动，下次还会踩。 # 所以纠正必须留一条看得见的痕迹，而不是安静地把事办了 ctx.emit(EventType.LOG, level="warn", - message=f"工具 {name}：{note}。请到节点里改正。") + message=f"工具 {name}：{note}。请到节点里改正。", code="tool_args_fixed") try: async with SessionLocal() as session: @@ -154,13 +154,15 @@ async def run_code(state: GraphState, ctx: NodeContext) -> dict[str, Any]: if isolation == "strict" and result.backend != "microvm": # 要了硬件隔离却没拿到，必须说出来——否则用户以为自己在 VM 里跑 ctx.emit(EventType.LOG, level="warn", - message=f"节点要求 strict 隔离，但 microVM 未就绪，实际用了 {result.backend}") + message=f"节点要求 strict 隔离，但 microVM 未就绪，实际用了 {result.backend}", + code="isolation_fallback") if result.session_reset: # 会话工作区被清空了。同一会话的下游节点会突然读不到自己写的文件， # 不说一声的话那个 FileNotFoundError 根本无从追查 ctx.emit(EventType.LOG, level="warn", message="沙箱会话已重置，此前在该会话写入的文件已丢失" - "（多为超时重建，或同会话中各节点的内存限额不一致）") + "（多为超时重建，或同会话中各节点的内存限额不一致）", + code="sandbox_reset") ctx.emit( EventType.SANDBOX_END, ok=result.ok, diff --git a/backend/app/engine/review.py b/backend/app/engine/review.py new file mode 100644 index 00000000..b28f23fc --- /dev/null +++ b/backend/app/engine/review.py @@ -0,0 +1,337 @@ +from __future__ import annotations + +import json +from collections.abc import Iterable, Mapping +from dataclasses import dataclass +from typing import Any + +from app.core.events import EventType +from app.engine.issuance import extract_numbers, trace_numbers + +# -------------------------------------------------------------------------- +# 跑完之后的复核层。 +# +# 编排跑完只说明「流程走完了」，不说明「答得对」。库里能查到大量 status=succeeded +# 的运行，过程中检索降级了、工具报错了、agent 步数用满了，而交到用户手上的答案 +# 对此只字不提——最糟的一次（run dd9927e6）交出去的干脆是一句内部管道文案。 +# +# 出具体系（issuance.py）本来就是干这个的，但它挂在输出节点的可选契约上，还要配 +# 口径卡节点提供受控指标集；问数据页生成的图基本都是 input → agent → output， +# 走不到那条路。所以这里补一层轻的： +# +# 1. scan() 纯规则扫一遍事件流，零成本。干净的运行到此为止，不调模型。 +# 2. review() 只在扫出信号时调一次模型，重组答案并把异常说破。 +# 3. 复核模型看不到原始数据，所以它写出来的任何新数字都是凭空来的 +# —— check_numbers() 机械拦截，对不上就整条退回原答案。 +# -------------------------------------------------------------------------- + +BROKEN = "broken" # 答案不可信 +DEGRADED = "degraded" # 能用，但有缺口要说明 +BENIGN = "benign" # 记在轨迹里就够了，不值得为它叫一次模型 + +_SEVERITY_RANK = {BENIGN: 0, DEGRADED: 1, BROKEN: 2} + + +@dataclass(frozen=True) +class Signal: + """一条「这次运行有什么不对」。detail 同时给模型和用户看，所以写人话。""" + + kind: str + detail: str + severity: str = DEGRADED + + def as_dict(self) -> dict[str, str]: + return {"kind": self.kind, "detail": self.detail, "severity": self.severity} + + +# warn 日志按 code 分类，而不是靠匹配中文消息串——消息是给人读的，随时会改， +# 把复核的判断挂在它上面等于给自己埋一颗定时炸弹。各 emit 点带上 code。 +# +# 没有 code 的（老运行留在库里的事件）一律按 degraded 处理：宁可多说一句， +# 也好过把真问题当噪音吞掉。 +WARN_CODES: dict[str, tuple[str, str]] = { + # 对答案内容零影响，轨迹里有就够了 + "stream_fallback": ("stream_fallback", BENIGN), + # 安全属性上的降级，不是答案内容的问题；轨迹里已经说清楚了 + "isolation_fallback": ("isolation_fallback", BENIGN), + # 答案本身就没写完 + "step_limit": ("step_limit", BROKEN), + "empty_completion": ("empty_completion", BROKEN), + # 能用但有缺口 + "loop_limit": ("loop_limit", DEGRADED), + "tool_args_fixed": ("tool_args_fixed", DEGRADED), + "node_retry": ("node_retry", DEGRADED), + "validate_retry": ("validate_retry", DEGRADED), + "sandbox_reset": ("sandbox_reset", DEGRADED), + "retrieve_degraded": ("retrieve_degraded", DEGRADED), +} + +# 这些 kind 重跑一次大概率能好：原因说得清，且不是「问题本身没问明白」。 +# 工具参数错了、步数不够、图没校验过，都属于改一下再来一遍就行的。 +RETRYABLE = {"step_limit", "tool_error", "empty_completion", "node_failed"} + +#: 串行工具模式下塞给未执行调用的占位说明。它漏到用户面前当过答案， +#: 所以复核要认得出来。从 nodes.llm 导入会绕成循环，这里按内容识别。 +_PLACEHOLDER_MARKS = ("本轮只执行了第一个工具",) + + +def answer_text(output: Mapping[str, Any] | None) -> str: + """把 output 各字段拼成答案正文。和前端 answerText 同一套规则。""" + if not output: + return "" + parts = [ + value if isinstance(value, str) else json.dumps(value, ensure_ascii=False) + for key, value in output.items() + if not key.startswith("_") and value not in (None, "") + ] + return "\n".join(parts).strip() + + +def _event_field(event: Any, name: str, default: Any = None) -> Any: + """事件既可能是 ORM 行也可能是 dict，两种都要吃得下。""" + if isinstance(event, Mapping): + return event.get(name, default) + return getattr(event, name, default) + + +def scan(events: Iterable[Any], output: Mapping[str, Any] | None = None) -> list[Signal]: + """纯规则扫一遍运行事件，挑出会影响答案可信度的那些。 + + 不调模型、不碰网络。返回空列表就意味着这次运行干净，可以直接交付。 + """ + signals: list[Signal] = [] + seen: set[tuple[str, str]] = set() + + def add(kind: str, detail: str, severity: str) -> None: + if severity == BENIGN: + return + key = (kind, detail) + if key in seen: + return + seen.add(key) + kinds.add(kind) + signals.append(Signal(kind, detail, severity)) + + kinds: set[str] = set() + + rows = list(events) + + # 工具报错之后，同一个工具在后面又跑成了，说明自愈接住了——那是缺口不是断裂。 + # 不区分的话，参数自动纠正这条路每修一次就报一次「答案不可信」。 + recovered: set[str] = set() + failed_at: dict[str, int] = {} + for i, e in enumerate(rows): + etype = str(_event_field(e, "type", "")) + data = _event_field(e, "data") or {} + tool = str(data.get("tool") or "") + if etype == EventType.TOOL_ERROR and tool: + failed_at.setdefault(tool, i) + elif etype == EventType.TOOL_END and tool and tool in failed_at: + recovered.add(tool) + + for e in rows: + etype = str(_event_field(e, "type", "")) + data = _event_field(e, "data") or {} + node_id = _event_field(e, "node_id") or "" + + if etype == EventType.TOOL_ERROR: + tool = str(data.get("tool") or "工具") + err = str(data.get("error") or "").strip()[:200] + if tool in recovered: + add("tool_args_fixed", f"工具 {tool} 第一次调用失败（{err}），重试后成功", DEGRADED) + else: + add("tool_error", f"工具 {tool} 调用失败：{err}", BROKEN) + + elif etype == EventType.NODE_FAILED: + err = str(data.get("error") or "").strip()[:200] + add("node_failed", f"节点 {node_id} 执行失败：{err}", BROKEN) + + elif etype == EventType.RUN_FAILED: + err = str(data.get("error") or "").strip()[:200] + add("run_failed", f"运行失败：{err}", BROKEN) + + elif etype == EventType.RETRIEVE_END: + if data.get("degraded") and "retrieve_degraded" not in kinds: + # knowledge.py 在这条事件**之前**已经把每条具体原因发成了 warn + #（"重排结果解析不出来"、"多少条向量对不上"）。两边都收就是同一件事 + # 报两遍，一条含糊一条具体——这里只兜没有 warn 的情形 + add("retrieve_degraded", "知识库检索能力降级，可能漏掉换了说法的内容", DEGRADED) + if not data.get("count"): + q = str(data.get("query") or "").strip()[:60] + add("retrieve_empty", f"知识库里没有检索到与「{q}」相关的内容", BROKEN) + + elif etype == EventType.LOG and data.get("level") == "warn": + message = str(data.get("message") or "").strip() + code = str(data.get("code") or "") + kind, severity = WARN_CODES.get(code, ("warn", DEGRADED)) + if message: + add(kind, message[:300], severity) + + text = answer_text(output) + if not text: + add("empty_output", "这次运行没有产出任何成果内容", BROKEN) + elif any(mark in text for mark in _PLACEHOLDER_MARKS): + add("placeholder_output", "交出来的是一句内部流程说明，不是对问题的回答", BROKEN) + + return signals + + +def worst(signals: Iterable[Signal]) -> str: + """一组信号里最重的那档。没有信号时返回空串。""" + ranked = sorted(signals, key=lambda s: _SEVERITY_RANK.get(s.severity, 1), reverse=True) + return ranked[0].severity if ranked else "" + + +def should_retry(signals: Iterable[Signal]) -> bool: + """这些信号里有没有「重跑一次大概率能好」的。""" + return any(s.kind in RETRYABLE and s.severity == BROKEN for s in signals) + + +def check_numbers(original: str, revised: str) -> list[str]: + """复核重写的答案里，有哪些数字是原答案里没有的。 + + 复核模型手上只有原答案和一份异常清单，没有任何原始数据——它写出来的新数字 + 只能是凭空来的。这条是硬拦截，不是提示：叙述层无算术权限是这个项目的地基 + （见 issuance.decide_tier）。 + + 直接复用出具校验那套回指逻辑，把原答案的数字当成受控指标集。 + """ + controlled = [ + {"id": tok.raw, "value": tok.value} for tok in extract_numbers(original) + ] + report = trace_numbers(revised, controlled) + return [item["token"] for item in report.unmatched] + + +REVIEW_SYSTEM = ( + "你负责复核一次数据问答的结果。执行过程中检测到了异常，你的任务是让用户" + "在读这个答案之前就知道它哪里不可靠。\n\n" + "你会拿到：用户的问题、系统给出的原答案、这次运行的异常清单。\n\n" + "铁律：\n" + "1. 你看不到任何原始数据。**不许写出原答案里没有的数字**——一个都不行。" + "需要改写时只能重组原答案已有的内容。\n" + "2. 异常必须写进 note，不许淡化或略过。降级要明说，不能悄悄地降。\n" + "3. 如果异常说明这个答案根本不可信（比如取数工具就没跑通、检索什么都没查到），" + "就直说答不了、说清缺什么，不要拿一个看起来完整的答案去糊。\n" + "4. note 写给用户看，一到两句中文，说人话：出了什么事、对这个答案意味着什么。" + "不要复述事件名或代码。\n\n" + "输出 JSON：\n" + '- note：给用户的异常说明（必填）\n' + '- answer：重写后的答案。原答案本身没问题、只是需要补一句说明时，留空字符串\n' + '- retry：这次运行是不是「换个配置重跑一遍大概率就好了」（比如步数不够、' + "工具参数写错）。问题本身没问清楚导致的，填 false" +) + +REVIEW_SCHEMA: dict[str, Any] = { + "title": "review", + "type": "object", + "properties": { + "note": {"type": "string", "description": "给用户的异常说明，一到两句中文"}, + "answer": {"type": "string", "description": "重写后的答案；不需要改写就留空"}, + "retry": {"type": "boolean", "description": "重跑一次是否大概率能好"}, + }, + "required": ["note"], +} + + +def review_prompt(question: str, answer: str, signals: list[Signal]) -> str: + """拼给复核模型的请求。 + + 只给信号清单，不给原始事件流：几百条事件塞进 prompt 既贵又没用，而信号 + 已经是规则层提炼过的结论。 + """ + lines = "\n".join( + f"- [{'严重' if s.severity == BROKEN else '降级'}] {s.detail}" for s in signals + ) + return ( + f"用户的问题：\n{question or '（未记录）'}\n\n" + f"系统给出的原答案：\n{answer or '（空）'}\n\n" + f"检测到的异常：\n{lines}" + ) + + +@dataclass +class ReviewResult: + """复核结论。verdict 是对「系统实际做了什么」的陈述，不是模型的自我评价。""" + + verdict: str # ok | annotated | rewritten | reverted | failed + note: str = "" + answer: str | None = None # None 表示沿用原答案 + #: 改写之前的那份。改写是有损的，原件得跟着一起落库——否则刷新之后 + #: 只剩一个被重组过的答案，想对照都没得对照 + original: str | None = None + retry: bool = False + signals: list[Signal] = None # type: ignore[assignment] + severity: str = "" + + def as_dict(self) -> dict[str, Any]: + return { + "verdict": self.verdict, + "note": self.note, + "answer": self.answer, + "original": self.original, + "retry": self.retry, + "severity": self.severity, + "signals": [s.as_dict() for s in (self.signals or [])], + } + + +async def review( + model: Any, *, question: str, answer: str, signals: list[Signal] +) -> ReviewResult: + """叫一次模型复核。调用方负责确保 signals 非空——空信号根本不该走到这里。 + + 复核本身失败不能让整轮挂掉：答案已经在手上了，复核只是加一层说明。 + 所以这里把异常全兜住，最差也要把原答案原样交出去。 + """ + severity = worst(signals) + rule_retry = should_retry(signals) + fallback_note = "；".join(s.detail for s in signals if s.severity == BROKEN) or "；".join( + s.detail for s in signals + ) + + messages = [("system", REVIEW_SYSTEM), ("human", review_prompt(question, answer, signals))] + try: + raw = await model.with_structured_output(REVIEW_SCHEMA).ainvoke(messages) + if not isinstance(raw, dict): + raise TypeError("结构化输出没有返回对象") + except Exception: # noqa: BLE001 - 不支持结构化输出的模型退回文本解析 + try: + from app.api.copilot import _extract_json + from app.engine.state import message_text + + raw = _extract_json(message_text(await model.ainvoke(messages))) + except Exception: # noqa: BLE001 + # 模型这条路彻底不通。规则层已经知道出了什么事，照样说得出话 + return ReviewResult( + verdict="annotated", note=fallback_note, retry=rule_retry, + signals=signals, severity=severity, + ) + + note = str(raw.get("note") or "").strip() or fallback_note + revised = str(raw.get("answer") or "").strip() + # 规则层管**资格**，模型只有**否决权**。反过来（两边取并）的话，模型就能 + # 越过规则强推重跑——而"检索什么都没查到"这类，重跑一遍还是查不到， + # 白烧一次建图加一次跑图。模型没表态时按规则走 + retry = rule_retry and bool(raw.get("retry", True)) + + if not revised or revised == answer.strip(): + return ReviewResult( + verdict="annotated", note=note, retry=retry, signals=signals, severity=severity, + ) + + invented = check_numbers(answer, revised) + if invented: + # 重写里冒出了原答案没有的数字。整条退回——说明还是要给的， + # 但那份正文一个字都不能用 + return ReviewResult( + verdict="reverted", + note=f"{note}\n\n（复核改写里出现了原答案中没有的数字" + f"（{'、'.join(invented[:5])}），已退回原答案。）", + retry=retry, signals=signals, severity=severity, + ) + + return ReviewResult( + verdict="rewritten", note=note, answer=revised, original=answer, retry=retry, + signals=signals, severity=severity, + ) diff --git a/backend/tests/test_conversations.py b/backend/tests/test_conversations.py index fb041e01..84d1209a 100644 --- a/backend/tests/test_conversations.py +++ b/backend/tests/test_conversations.py @@ -157,6 +157,26 @@ async def test_history_text_is_plain_qa(client) -> None: assert "shop 里有几家店" in text and "一共 12 家" in text +async def test_history_carries_the_review_warning_forward(client) -> None: + """被复核判过不可信的答案，下一轮必须知道。 + + 只把正文喂过去的话，模型会拿一个已知有问题的结论当事实继续往下推 + （"那 12 家里有几家是新开的？"），错误就这么一轮轮传下去，而且越传 + 越像真的——后面几轮看起来都在正常作答。 + """ + conv = await _conversation(client) + await _turn( + client, conv["id"], "shop 里有几家店", + answer="一共 12 家", status="done", + review={"verdict": "annotated", "severity": "broken", "note": "取数没跑通，这个数不可信。", + "answer": None, "retry": True, "signals": []}, + ) + async with SessionLocal() as session: + text = await history_text(session, conv["id"]) + assert "一共 12 家" in text + assert "取数没跑通" in text + + async def test_history_text_is_empty_without_a_conversation(client) -> None: async with SessionLocal() as session: assert await history_text(session, None) == "" diff --git a/backend/tests/test_review.py b/backend/tests/test_review.py new file mode 100644 index 00000000..a58021aa --- /dev/null +++ b/backend/tests/test_review.py @@ -0,0 +1,363 @@ +"""跑完之后的复核层。 + +核心不变量有两条，其余都是围着它们转的： + +1. 干净的运行**一次模型都不调**。复核是补救手段，不是每次问数的固定开销。 +2. 复核模型看不到原始数据，所以它写出来的任何新数字都是现编的，必须机械拦住。 +""" + +from __future__ import annotations + +import pytest +from httpx import ASGITransport, AsyncClient + +from app.core.events import EventType +from app.engine import review as rv +from app.main import app + + +@pytest.fixture +async def client(): + transport = ASGITransport(app=app) + async with AsyncClient(transport=transport, base_url="http://test") as c: + yield c + + +def ev(etype, **data): + return {"type": str(etype), "node_id": data.pop("node_id", ""), "data": data} + + +def warn(message, code=""): + return ev(EventType.LOG, level="warn", message=message, code=code) + + +OK_OUTPUT = {"结果": "管理员有 6 个。"} + + +# -------------------------------------------------------------------------- +# scan：规则层 +# -------------------------------------------------------------------------- + + +def test_clean_run_has_no_signals(): + """干净的运行扫不出东西——这是"零成本"的前提。""" + events = [ + ev(EventType.RUN_STARTED), + ev(EventType.TOOL_START, tool="db_query__shop"), + ev(EventType.TOOL_END, tool="db_query__shop"), + ev(EventType.LOG, level="info", message="随便什么"), + ev(EventType.RETRIEVE_END, count=5, degraded=False, query="权限"), + ev(EventType.RUN_FINISHED), + ] + assert rv.scan(events, OK_OUTPUT) == [] + + +def test_tool_error_is_broken(): + signals = rv.scan([ev(EventType.TOOL_ERROR, tool="db_query__x", error="no such table")], + OK_OUTPUT) + assert [s.kind for s in signals] == ["tool_error"] + assert signals[0].severity == rv.BROKEN + assert "no such table" in signals[0].detail + + +def test_tool_error_then_success_is_only_degraded(): + """自愈接住了就不该报"答案不可信"。 + + 参数自动纠正那条路每修一次就发一条 tool.error，不区分的话每次自愈 + 都会触发一次复核，还告诉用户答案不可信——而它明明跑通了。 + """ + signals = rv.scan([ + ev(EventType.TOOL_ERROR, tool="db_query__x", error="unexpected keyword 'query'"), + ev(EventType.TOOL_END, tool="db_query__x"), + ], OK_OUTPUT) + assert [s.kind for s in signals] == ["tool_args_fixed"] + assert signals[0].severity == rv.DEGRADED + + +def test_retrieve_degraded_and_empty(): + degraded = rv.scan([ev(EventType.RETRIEVE_END, count=3, degraded=True)], OK_OUTPUT) + assert degraded[0].kind == "retrieve_degraded" + assert degraded[0].severity == rv.DEGRADED + + empty = rv.scan([ev(EventType.RETRIEVE_END, count=0, query="商家账户")], OK_OUTPUT) + assert empty[0].kind == "retrieve_empty" + assert empty[0].severity == rv.BROKEN + assert "商家账户" in empty[0].detail + + +def test_specific_retrieve_reason_wins_over_the_generic_one(): + """检索降级会发两条：knowledge.py 先发具体原因的 warn，再发 retrieve.end。 + + 两边都收就是同一件事报两遍，用户读到一条含糊的加一条具体的，还得自己 + 判断是不是两个问题。 + """ + signals = rv.scan([ + warn("default 里有 32 条向量由「local-hashing」建立，和当前的对不上", "retrieve_degraded"), + ev(EventType.RETRIEVE_END, count=5, degraded=True), + ], OK_OUTPUT) + assert len(signals) == 1 + assert "32 条向量" in signals[0].detail + + +def test_node_and_run_failures_are_broken(): + signals = rv.scan([ + ev(EventType.NODE_FAILED, node_id="agent1", error="boom"), + ev(EventType.RUN_FAILED, error="超时"), + ], OK_OUTPUT) + assert {s.kind for s in signals} == {"node_failed", "run_failed"} + assert all(s.severity == rv.BROKEN for s in signals) + + +def test_warn_classified_by_code_not_by_message(): + """分类挂在 code 上。消息是给人读的，随时会改。""" + assert rv.scan([warn("agent 用满了 8 步还没给出结论。", "step_limit")], OK_OUTPUT)[0] \ + .severity == rv.BROKEN + assert rv.scan([warn("第 1 次校验失败：不是合法 JSON", "validate_retry")], OK_OUTPUT)[0] \ + .severity == rv.DEGRADED + + +def test_benign_warns_do_not_trigger_review(): + """流式回退、隔离降级对答案内容零影响。 + + 为它们叫一次模型，等于每次网络抖一下就多花一次调用，还要用户读一句 + 与结论无关的告警。轨迹里有记录就够了。 + """ + events = [ + warn("流式失败，回退非流式：connection reset", "stream_fallback"), + warn("节点要求 strict 隔离，但 microVM 未就绪，实际用了 seatbelt", "isolation_fallback"), + ] + assert rv.scan(events, OK_OUTPUT) == [] + + +def test_warn_without_code_defaults_to_degraded(): + """库里的老事件没有 code。宁可多说一句，也好过把真问题当噪音吞掉。""" + signals = rv.scan([warn("某个没带 code 的告警")], OK_OUTPUT) + assert [s.kind for s in signals] == ["warn"] + assert signals[0].severity == rv.DEGRADED + + +def test_empty_and_placeholder_output_are_broken(): + assert rv.scan([], None)[0].kind == "empty_output" + assert rv.scan([], {"结果": ""})[0].kind == "empty_output" + + # run dd9927e6 真的把这句内部管道文案当答案交出去过 + leaked = {"结果": "本轮只执行了第一个工具。看到它的结果之后，再决定这一个还要不要调。"} + assert rv.scan([], leaked)[0].kind == "placeholder_output" + + +def test_duplicate_signals_collapse(): + """一次运行里同一条告警发五遍，用户只需要看到一次。""" + events = [warn("第 1 次校验失败：不是合法 JSON", "validate_retry")] * 5 + assert len(rv.scan(events, OK_OUTPUT)) == 1 + + +def test_scan_accepts_orm_like_rows(): + """事件既可能是 dict 也可能是 ORM 行，两种都要吃得下。""" + + class Row: + type = str(EventType.TOOL_ERROR) + node_id = "n1" + data = {"tool": "t", "error": "e"} + + assert rv.scan([Row()], OK_OUTPUT)[0].kind == "tool_error" + + +def test_worst_and_should_retry(): + broken = rv.Signal("step_limit", "步数用满", rv.BROKEN) + degraded = rv.Signal("retrieve_degraded", "检索降级", rv.DEGRADED) + assert rv.worst([degraded, broken]) == rv.BROKEN + assert rv.worst([degraded]) == rv.DEGRADED + assert rv.worst([]) == "" + + assert rv.should_retry([broken]) is True + # 检索什么都没查到，重跑一遍还是查不到——那不是重跑能解决的 + assert rv.should_retry([rv.Signal("retrieve_empty", "没命中", rv.BROKEN)]) is False + assert rv.should_retry([degraded]) is False + + +# -------------------------------------------------------------------------- +# check_numbers：不得新增数字 +# -------------------------------------------------------------------------- + + +def test_check_numbers_allows_reordering(): + original = "管理员有 6 个，商家有 5 个，占比 45.5%。" + revised = "占比 45.5%。其中商家 5 个、管理员 6 个。" + assert rv.check_numbers(original, revised) == [] + + +def test_check_numbers_catches_invented(): + invented = rv.check_numbers("管理员有 6 个。", "管理员有 6 个，商家有 5 个。") + assert invented == ["5"] + + +def test_check_numbers_ignores_prose_without_numbers(): + assert rv.check_numbers("管理员有 6 个。", "这次没能查到有效数据，建议重试。") == [] + + +# -------------------------------------------------------------------------- +# review：调模型那一段 +# -------------------------------------------------------------------------- + + +class FakeModel: + """最小的 with_structured_output 替身。calls 记录被调了几次。""" + + def __init__(self, reply): + self.reply = reply + self.calls = 0 + + def with_structured_output(self, schema): + return self + + async def ainvoke(self, messages): + self.calls += 1 + if isinstance(self.reply, Exception): + raise self.reply + return self.reply + + +SIGNALS = [rv.Signal("step_limit", "agent 用满了 8 步还没给出结论", rv.BROKEN)] + + +@pytest.mark.asyncio +async def test_review_annotates_without_rewriting(): + model = FakeModel({"note": "这次没跑完就收尾了。", "answer": "", "retry": True}) + result = await rv.review(model, question="有几个管理员", answer="管理员有 6 个。", + signals=SIGNALS) + assert result.verdict == "annotated" + assert result.answer is None # 沿用原答案 + assert result.note == "这次没跑完就收尾了。" + assert result.retry is True + assert result.severity == rv.BROKEN + + +@pytest.mark.asyncio +async def test_review_adopts_clean_rewrite(): + model = FakeModel({ + "note": "步数用满了。", "answer": "管理员有 6 个（这个数可能不完整）。", "retry": False, + }) + result = await rv.review(model, question="q", answer="管理员有 6 个。", signals=SIGNALS) + assert result.verdict == "rewritten" + assert result.answer == "管理员有 6 个（这个数可能不完整）。" + # 规则层管资格、模型有否决权：step_limit 本来够格重跑，模型说不必就不跑 + assert result.retry is False + + +@pytest.mark.asyncio +async def test_model_cannot_force_an_unhelpful_retry(): + """检索什么都没查到，重跑一遍还是查不到——模型说要重跑也不能听。 + + 否则就是白烧一次建图加一次跑图，而用户等的时间翻倍。 + """ + model = FakeModel({"note": "知识库里没有这个。", "answer": "", "retry": True}) + result = await rv.review( + model, question="q", answer="没查到。", + signals=[rv.Signal("retrieve_empty", "知识库里没有检索到相关内容", rv.BROKEN)], + ) + assert result.severity == rv.BROKEN + assert result.retry is False + + +@pytest.mark.asyncio +async def test_review_reverts_rewrite_with_invented_numbers(): + """复核手上只有原答案，它写出来的新数字只能是凭空来的。""" + model = FakeModel({ + "note": "步数用满了。", "answer": "管理员有 6 个，商家有 5 个。", "retry": False, + }) + result = await rv.review(model, question="q", answer="管理员有 6 个。", signals=SIGNALS) + assert result.verdict == "reverted" + assert result.answer is None # 那份正文一个字都不能用 + assert "5" in result.note # 但要说清是为什么退的 + assert "步数用满了。" in result.note # 说明本身还是要给 + + +@pytest.mark.asyncio +async def test_review_survives_model_failure(): + """复核只是加一层说明，它自己挂了不能把已经到手的答案也带走。""" + model = FakeModel(RuntimeError("502")) + result = await rv.review(model, question="q", answer="管理员有 6 个。", signals=SIGNALS) + assert result.verdict == "annotated" + assert result.answer is None + assert result.note == SIGNALS[0].detail # 退回规则层自己的措辞 + assert result.retry is True + + +@pytest.mark.asyncio +async def test_review_survives_garbage_reply(): + model = FakeModel("这不是 JSON，也不是对象") + result = await rv.review(model, question="q", answer="管理员有 6 个。", signals=SIGNALS) + assert result.verdict == "annotated" + assert result.note + + +# -------------------------------------------------------------------------- +# 接口 +# -------------------------------------------------------------------------- + + +async def _make_run(client, output, events): + """直接落一条运行和它的事件，省掉真跑一遍图。""" + from app.db.base import SessionLocal + from app.db.models import Run, RunEvent + + async with SessionLocal() as session: + run = Run(workflow_id=None, status="succeeded", output=output, graph={}) + session.add(run) + await session.flush() + for i, e in enumerate(events): + session.add(RunEvent( + run_id=run.id, seq=i + 1, type=e["type"], + node_id=e["node_id"] or None, ts=0.0, data=e["data"], + )) + await session.commit() + return run.id + + +@pytest.mark.asyncio +async def test_clean_run_never_touches_the_model(client, monkeypatch): + """干净运行零成本——这条是整层设计的前提，要用「模型根本没被碰过」来断言。""" + async def explode(*a, **k): + raise AssertionError("干净的运行不该调模型") + + monkeypatch.setattr("app.providers.factory.get_chat_model", explode) + monkeypatch.setattr("app.api.copilot.get_chat_model", explode) + + run_id = await _make_run(client, OK_OUTPUT, [ + ev(EventType.TOOL_START, tool="t"), ev(EventType.TOOL_END, tool="t"), + ]) + resp = await client.post("/api/copilot/review", json={"run_id": run_id, "question": "q"}) + assert resp.status_code == 200 + body = resp.json() + assert body["verdict"] == "ok" + assert body["signals"] == [] + assert body["answer"] is None + + +@pytest.mark.asyncio +async def test_review_endpoint_reports_signals(client, monkeypatch): + """没配模型时也得把规则层扫到的东西说出来，而不是静默放行。""" + from app.providers.factory import ProviderNotConfigured + + async def unconfigured(*a, **k): + raise ProviderNotConfigured("没有可用的 provider") + + monkeypatch.setattr("app.api.copilot.get_chat_model", unconfigured) + + run_id = await _make_run(client, OK_OUTPUT, [ + ev(EventType.TOOL_ERROR, tool="db_query__x", error="no such table: users"), + ]) + body = (await client.post( + "/api/copilot/review", json={"run_id": run_id, "question": "q"} + )).json() + assert body["verdict"] == "annotated" + assert body["severity"] == rv.BROKEN + assert [s["kind"] for s in body["signals"]] == ["tool_error"] + assert "no such table" in body["note"] + assert body["retry"] is True + + +@pytest.mark.asyncio +async def test_review_endpoint_404(client): + resp = await client.post("/api/copilot/review", json={"run_id": "没这个", "question": "q"}) + assert resp.status_code == 404 diff --git a/frontend/src/api/client.ts b/frontend/src/api/client.ts index ef383964..f4f0f3d6 100644 --- a/frontend/src/api/client.ts +++ b/frontend/src/api/client.ts @@ -1,6 +1,6 @@ import type { Approval, Conversation, ConversationDetail, ConversationTurn, GraphSpec, - KbDocument, MemoryItem, Provider, Run, RunEvent, Skill, ToolInfo, + KbDocument, MemoryItem, Provider, ReviewResult, Run, RunEvent, Skill, ToolInfo, ValidationIssue, VarIssue, Variable, Workflow, } from '../types' @@ -258,6 +258,9 @@ export const api = { }>('/copilot/model'), setModel: (body: { provider?: string | null; model?: string | null }) => put('/copilot/model', body), + /** 跑完之后复核一次。干净的运行后端直接返回 verdict='ok'，不调模型 */ + review: (body: { run_id: string; question: string }) => + post('/copilot/review', body), }, } diff --git a/frontend/src/dev/preview.tsx b/frontend/src/dev/preview.tsx index 6f66c34c..8d7ad23e 100644 --- a/frontend/src/dev/preview.tsx +++ b/frontend/src/dev/preview.tsx @@ -102,6 +102,40 @@ const NO_QUERY_TURN: StreamTurn = { output: { answer: "第 1 轮走的是 `role.level IN (platform, regional)` 这个口径。要换口径我重新查一遍。" }, } +/** 复核判了「不可信」：说明排在成果上方，且用报警色 */ +const REVIEW_BROKEN_TURN: StreamTurn = { + id: "reviewbroken", question: "商家账户有多少个？", phase: "done", status: "完成", + steps: [], + review: { + verdict: "annotated", severity: "broken", retry: true, + note: "这次取数没跑通（表名对不上），下面这个数是模型根据上下文推断的，不能当结论用。", + answer: null, + signals: [ + { kind: "tool_error", severity: "broken", + detail: "工具 db_query__shop 调用失败：no such table: merchant_account" }, + { kind: "step_limit", severity: "broken", detail: "agent 用满了 8 步还没给出结论" }, + ], + }, + output: { answer: "大约 5 个左右。" }, +} + +/** 复核只是补了一句说明，答案照常给；外加改写前的原文可展开 */ +const REVIEW_DEGRADED_TURN: StreamTurn = { + id: "reviewdegraded", question: "谁能改别人的权限？", phase: "done", status: "完成", + steps: [], + review: { + verdict: "rewritten", severity: "degraded", retry: false, + note: "这次知识库检索退回了纯关键词匹配，换了说法的内容可能没被搜到。", + answer: null, + signals: [ + { kind: "retrieve_degraded", severity: "degraded", + detail: "重排结果解析不出来，按初筛顺序返回" }, + ], + }, + output: { answer: "只有平台管理员可以修改他人权限，且需要走角色变更流程。" }, + rawOutput: { answer: "平台管理员可以修改他人权限。" }, +} + const LONG_TURN: StreamTurn = { id: "long", question: "需要能够显示总结内容", phase: "done", status: "完成", steps: [], output: { result: longReport() }, @@ -148,6 +182,9 @@ function Preview() { if (params.get('noquery') === '1') { return } + if (params.get('review') === '1') { + return + } return (
diff --git a/frontend/src/pages/ChatPage.tsx b/frontend/src/pages/ChatPage.tsx index 16e9dfb4..0c5a8fa9 100644 --- a/frontend/src/pages/ChatPage.tsx +++ b/frontend/src/pages/ChatPage.tsx @@ -149,6 +149,8 @@ function toStreamTurn(turn: ChatTurn): StreamTurn { graph: turn.graph, graphNote: turn.explanation, noQuery: turn.noQuery, + review: turn.review, + rawOutput: turn.rawOutput, } } diff --git a/frontend/src/run/AssistantStream.tsx b/frontend/src/run/AssistantStream.tsx index ac478e17..402cc8b1 100644 --- a/frontend/src/run/AssistantStream.tsx +++ b/frontend/src/run/AssistantStream.tsx @@ -1,6 +1,6 @@ import { useEffect, useRef, useState, type ReactNode } from 'react' import { - Brain, ChevronRight, CircleCheck, CircleDot, Database, + AlertTriangle, Brain, ChevronRight, CircleCheck, CircleDot, Database, FileCode, GitBranch, Hand, Info, Sparkles, Table2, Terminal, Wrench, XCircle, } from 'lucide-react' import clsx from 'clsx' @@ -9,6 +9,7 @@ import { parseQueryResult, type ResultTable as Table, type Step, type StepKind, } from './decode' import { Markdown } from './Markdown' +import type { ReviewResult } from '../types' /** * 助手流：一轮轮「你问什么 → 它做了什么 → 结果」。 @@ -42,6 +43,10 @@ export interface StreamTurn { graphNote?: string /** 这一条没有查库，答案是根据前几轮说的 */ noQuery?: boolean + /** 复核结论：这次运行有什么不对，以及对这个答案意味着什么 */ + review?: ReviewResult | null + /** 被复核重写之前的答案。改写是有损的，得能对照原件 */ + rawOutput?: Record | null } const ICONS: Record = { @@ -156,10 +161,25 @@ function TurnCard({ turn, dense, approvals, onOpenGraph }: {
)} + {/* 复核说明排在成果**上方**，和上面那条声明同一个位置。排在下面 + 等于让人读完整个结论、信了，才知道它是在什么条件下得出的 */} + {turn.review && } + {turn.output && ( )} + {turn.rawOutput && ( +
+ + 复核改写过这个答案，看改写前的原文 + +
+ +
+
+ )} + {turn.graph && ( @@ -169,6 +189,41 @@ function TurnCard({ turn, dense, approvals, onOpenGraph }: { ) } +/** + * 这次运行有什么不对。 + * + * broken 用报警色：那意味着答案本身不可信，不是"仅供参考"。degraded 用普通 + * 边框——它能用，只是有缺口。两者混成一个样子，用户就学会了一概忽略。 + */ +function ReviewNote({ review }: { review: ReviewResult }) { + const broken = review.severity === 'broken' + const Icon = broken ? AlertTriangle : Info + return ( +
+
+ + {review.note} +
+ {!!review.signals.length && ( +
+ + 检测到 {review.signals.length} 处异常 + +
    + {review.signals.map((s, i) => ( +
  • · {s.detail}
    • + ))} +
+
+ )} +
+ ) +} + /** 思考的尾窗：只露最后几行并跟着滚，不然一大坨会把步骤挤到屏幕外。 */ function ThinkingTail({ text }: { text: string }) { const ref = useRef(null) diff --git a/frontend/src/store/chat.ts b/frontend/src/store/chat.ts index 7add8c33..6dae3565 100644 --- a/frontend/src/store/chat.ts +++ b/frontend/src/store/chat.ts @@ -3,7 +3,7 @@ import { api, streamCopilot, streamRun } from '../api/client' import type { CopilotOp } from '../run/decode' import { decodeRun } from '../run/decode' import { useConversations } from './conversations' -import type { GraphSpec, Run, RunEvent } from '../types' +import type { GraphSpec, ReviewResult, Run, RunEvent, ValidationIssue } from '../types' /** * 对话式入口的状态。 @@ -70,6 +70,13 @@ export interface ChatTurn { restored?: boolean /** 已经补取过事件（或正在取），别重复拉 */ stepsLoaded?: boolean + /** 跑完之后的复核结论。null = 没复核过，和「复核过、没问题」不是一回事 */ + review?: ReviewResult | null + /** 被复核重写之前的答案。改写是有损的，原件得留着让用户能对照 */ + rawOutput?: Record | null + /** 这一轮重建过几次图。上限 1——第二次还不行多半是问题本身没说清， + * 该让用户介入，而不是替他一遍遍烧钱 */ + attempt?: number } interface ChatState { @@ -112,6 +119,9 @@ const EMPTY: ChatTurn[] = [] /** 落库成果的上限。只是防跑飞，不是显示上限 */ const ANSWER_CAP = 20_000 +/** 等复核的上限。超了就先把原答案交出去，说明后补不了就算了 */ +const REVIEW_TIMEOUT_MS = 25_000 + type Patch = (fn: (t: ChatTurn) => Partial) => void /** @@ -183,6 +193,10 @@ export const useChat = create((set, get) => ({ cancel: null, restored: true, stepsLoaded: false, + // 复核结论是答案的一部分。刷新之后只剩一个看起来很完整的答案、 + // 而"它哪里不可靠"没了，比一开始就不复核更糟 + review: t.review && t.review.verdict !== 'ok' ? (t.review as ReviewResult) : null, + rawOutput: t.review?.original ? { answer: t.review.original } : null, })) set((s) => ({ byConversation: { ...s.byConversation, [conversationId]: turns }, @@ -222,7 +236,7 @@ export const useChat = create((set, get) => ({ id, serverId: null, question, phase: 'planning', status: PHASE_TEXT.planning, thinking: '', ops: [], events: [], graph: null, explanation: '', run: null, output: null, error: '', startedAt: Date.now(), - lastSeq: 0, cancel: null, + lastSeq: 0, cancel: null, attempt: 0, }, ], }, @@ -244,94 +258,11 @@ export const useChat = create((set, get) => ({ } catch { /* 存不下不该影响正在进行的对话 */ } } - const cancelCopilot = streamCopilot( - { instruction: question, base_graph: null, intent: 'answer', conversation_id: conversationId }, - (op) => { - // 先原样收下，翻译交给 decodeCopilot。thinking 合并在写入时做： - // 一次生成几百上千条 delta，逐条存下来光数组就比图大一个量级 - patch((t) => { - const last = t.ops[t.ops.length - 1] - if (op.op === 'thinking' && last?.op === 'thinking') { - const merged = { ...last, delta: String(last.delta ?? '') + String(op.delta ?? '') } - return { ops: [...t.ops.slice(0, -1), merged] } - } - return { ops: [...t.ops, op] } - }) - - switch (op.op) { - case 'thinking': - patch((t) => ({ - phase: 'planning', - thinking: (t.thinking + (op.delta ?? '')).slice(-2000), - })) - break - case 'heartbeat': - patch(() => ({ status: PHASE_TEXT[(op.phase as Phase) ?? 'planning'] ?? '正在处理…' })) - break - case 'plan': - patch(() => ({ phase: 'building', status: op.summary || '正在搭建流程…' })) - break - case 'add_node': - patch((t) => ({ - phase: 'building', - status: `正在搭建流程…（${t.ops.filter((o) => o.op === 'add_node').length} 步）`, - })) - break - case 'done': - patch(() => ({ explanation: op.explanation ?? '' })) - break - case 'reply': { - // 这句话不需要工作流——问的是前几轮已经查出来的东西，或者是 - // 对过程说的话（「重试」「换个说法」）。以前这些也各自重建一整张 - // 图、跑一遍 153 张表的库，纯属白花钱 - const text = String(op.text ?? '') - patch(() => ({ - phase: 'done', status: PHASE_TEXT.done, - output: { answer: text }, noQuery: true, - })) - set({ busy: false, cancel: null }) - void save({ status: 'done', answer: text }) - break - } - case 'final': { - const graph = op.graph as GraphSpec - void save({ graph, explanation: op.explanation ?? '' }) - // 用户要的是流程本身时（「设计一个每天跑的工作流」），建完就跑 - // 等于替他多花一次钱，而他要的是那张图 - if (op.autorun === false) { - patch(() => ({ - graph, phase: 'ready', status: PHASE_TEXT.ready, - explanation: op.explanation ?? '', pendingRun: true, - })) - set({ busy: false, cancel: null }) - void save({ status: 'done' }) - break - } - patch(() => ({ - graph, phase: 'running', status: PHASE_TEXT.running, - explanation: op.explanation ?? '', - })) - void launch(conversationId, id, graph, question, patch, set, save) - break - } - case 'error': - patch(() => ({ phase: 'error', error: op.message ?? '生成失败' })) - set({ busy: false, cancel: null }) - void save({ status: 'error', error: op.message ?? '生成失败' }) - break - } - }, - (error) => { - if (!error) return - patch((t) => (t.phase === 'error' ? {} : { phase: 'error', error })) - set({ busy: false, cancel: null }) - void save({ status: 'error', error }) - }, - ) - patch(() => ({ cancel: cancelCopilot })) - set({ cancel: cancelCopilot }) + plan({ + conversationId, turnId: id, question, instruction: question, + attempt: 0, patch, set, save, + }) }, - stop: (conversationId) => { // 只停最后一轮。把所有非终态 turn 一律标成"已取消"会误杀正等人工介入的 // 历史轮次——那些是等着用户回来处理的，不是卡住的。 @@ -385,7 +316,7 @@ export const useChat = create((set, get) => ({ } patch(() => ({ phase: 'running', status: PHASE_TEXT.running })) set({ busy: true }) - watch(turn.run.id, patch, set, save, turn.lastSeq) + watch(turn.run.id, patch, set, save, turn.lastSeq, turn.question) }, })) @@ -450,9 +381,187 @@ function historyOf(conversationId: string, upToTurnId: string): string { } /** 图建好了就直接跑——用户要的是答案，不是一张图。 */ +/** 一轮最多再试一次。第二次还不行多半是问题本身没说清，该让用户介入 */ +const MAX_ATTEMPTS = 2 + +interface PlanArgs { + conversationId: string + turnId: string + /** 用户原本问的是什么。跑图取数、复核都要用它，不能被重试的补充说明污染 */ + question: string + /** 真正发给 Copilot 的那句话。重试时会在后面缀上失败原因 */ + instruction: string + /** 重试时把上一次那张图交回去，让它改而不是重建。null = 从零建 */ + baseGraph?: GraphSpec | null + attempt: number + patch: Patch + set: any + save: (body: Record) => Promise +} + +/** + * 建图 → 跑图。从 ask 里拆出来，是为了能带着失败原因原地重来一次。 + * + * 重试必须回到**同一轮**：用户问了一个问题，中间重建了一次图是过程，不是 + * 两次对话。之前这件事全靠用户自己发现不对、再打一句「重来」——他那句话 + * 本来就是在手动做这件事。 + */ +function plan(args: PlanArgs) { + const { + conversationId, turnId, question, instruction, baseGraph, attempt, patch, set, save, + } = args + + /** 这一轮建出来的图。retry 要把它交回去，所以得在闭包里留一份 */ + let built: GraphSpec | null = baseGraph ?? null + + /** + * 带着原因重来一次。超过上限就把原因如实留在这一轮上，不再烧钱。 + * + * 关键是**把失败的那张图一起交回去**。只给一句"上次没成功：工具参数错了" + * 而不给图，模型只能照着原问题从零再建一遍——很可能建出一模一样的那张， + * 白跑一个来回。而且它连自己上次建了什么都看不到：会话历史按 + * `status != running` 过滤，正在重试的这一轮恰好被排除在外。 + * + * 交回去之后走的是后端的"改图"分支：图预置进累加器，模型只输出改动操作， + * 于是调大最大步数、改正工具参数这种事就是改一个字段，不是推倒重来。 + */ + const retry = (reason: string) => { + if (attempt + 1 >= MAX_ATTEMPTS) return false + const base = built && (built.nodes?.length ?? 0) > 0 ? built : null + patch(() => ({ + attempt: attempt + 1, + phase: 'planning', + status: base ? '上一次没跑通，正在调整流程…' : '上一次没跑通，正在重新搭建流程…', + // 旧的操作流和事件要清掉：留着会让"执行过程"里出现两段互相矛盾的记录， + // 而用户分不清哪段是最终那次。图不清——它正是这次要改的东西 + ops: [], events: [], run: null, output: null, + review: null, rawOutput: null, lastSeq: 0, + })) + plan({ + ...args, + attempt: attempt + 1, + baseGraph: base, + instruction: base + ? `这张流程上一次运行没有成功：${reason}。\n\n` + + `请针对这个原因改它（比如调大最大步数、改正工具参数、换个查询方式），` + + `改完继续回答原来的问题：${question}` + : `${question}\n\n（上一次没有成功：${reason}。请针对这个原因调整，重新给出完整的工作流。）`, + }) + return true + } + + const cancelCopilot = streamCopilot( + { instruction, base_graph: baseGraph ?? null, intent: 'answer', conversation_id: conversationId }, + (op) => { + // 先原样收下，翻译交给 decodeCopilot。thinking 合并在写入时做： + // 一次生成几百上千条 delta，逐条存下来光数组就比图大一个量级 + patch((t) => { + const last = t.ops[t.ops.length - 1] + if (op.op === 'thinking' && last?.op === 'thinking') { + const merged = { ...last, delta: String(last.delta ?? '') + String(op.delta ?? '') } + return { ops: [...t.ops.slice(0, -1), merged] } + } + return { ops: [...t.ops, op] } + }) + switch (op.op) { + case 'thinking': + patch((t) => ({ + phase: 'planning', + thinking: (t.thinking + (op.delta ?? '')).slice(-2000), + })) + break + case 'heartbeat': + patch(() => ({ status: PHASE_TEXT[(op.phase as Phase) ?? 'planning'] ?? '正在处理…' })) + break + case 'plan': + patch(() => ({ phase: 'building', status: op.summary || '正在搭建流程…' })) + break + case 'add_node': + patch((t) => ({ + phase: 'building', + status: `正在搭建流程…（${t.ops.filter((o) => o.op === 'add_node').length} 步）`, + })) + break + case 'done': + patch(() => ({ explanation: op.explanation ?? '' })) + break + case 'reply': { + // 这句话不需要工作流——问的是前几轮已经查出来的东西，或者是 + // 对过程说的话（「重试」「换个说法」）。以前这些也各自重建一整张 + // 图、跑一遍 153 张表的库，纯属白花钱 + const text = String(op.text ?? '') + patch(() => ({ + phase: 'done', status: PHASE_TEXT.done, + output: { answer: text }, noQuery: true, + })) + set({ busy: false, cancel: null }) + void save({ status: 'done', answer: text }) + break + } + case 'final': { + const graph = op.graph as GraphSpec + built = graph // 要重试的话，交回去的就是它 + void save({ graph, explanation: op.explanation ?? '' }) + + // 没过校验的图不能拿去跑：引擎会照实抛出它自己的措辞（"图是空的， + // 先拖一个节点进来"），而用户根本不在画布上，这句话对他毫无意义。 + // 这条路真的把它当答案交到用户面前过 + const blockers = ((op.issues ?? []) as ValidationIssue[]) + .filter((i) => i.level === 'error') + .map((i) => i.message) + if (blockers.length && retry(`生成的工作流没有通过校验：${blockers.join('；')}`)) break + if (blockers.length) { + // 重来一次还是没过。校验消息是写给画布用户的（"先拖一个节点进来"）， + // 直接甩给一个在问数据页打字的人毫无意义——先给一句他能照做的， + // 原因放后面留作线索 + const why = `这次没能搭出可以运行的流程，你可以换个说法再问一次。` + + `（原因：${blockers.join('；')}）` + patch(() => ({ graph, phase: 'error', error: why })) + set({ busy: false, cancel: null }) + void save({ status: 'error', error: why }) + break + } + + // 用户要的是流程本身时（「设计一个每天跑的工作流」），建完就跑 + // 等于替他多花一次钱，而他要的是那张图 + if (op.autorun === false) { + patch(() => ({ + graph, phase: 'ready', status: PHASE_TEXT.ready, + explanation: op.explanation ?? '', pendingRun: true, + })) + set({ busy: false, cancel: null }) + void save({ status: 'done' }) + break + } + patch(() => ({ + graph, phase: 'running', status: PHASE_TEXT.running, + explanation: op.explanation ?? '', + })) + void launch(conversationId, turnId, graph, question, patch, set, save, retry) + break + } + case 'error': + patch(() => ({ phase: 'error', error: op.message ?? '生成失败' })) + set({ busy: false, cancel: null }) + void save({ status: 'error', error: op.message ?? '生成失败' }) + break + } + }, + (error) => { + if (!error) return + patch((t) => (t.phase === 'error' ? {} : { phase: 'error', error })) + set({ busy: false, cancel: null }) + void save({ status: 'error', error }) + }, + ) + patch(() => ({ cancel: cancelCopilot })) + set({ cancel: cancelCopilot }) +} + async function launch( conversationId: string, turnId: string, graph: GraphSpec, question: string, patch: Patch, set: any, save: (body: Record) => Promise, + retry?: (reason: string) => boolean, ) { try { const run = await api.runs.start({ @@ -460,7 +569,7 @@ async function launch( }) patch(() => ({ run })) void save({ run_id: run.id }) - watch(run.id, patch, set, save) + watch(run.id, patch, set, save, 0, question, retry) } catch (e: any) { patch(() => ({ phase: 'error', error: e?.message ?? '启动失败' })) set({ busy: false, cancel: null }) @@ -477,6 +586,7 @@ async function launch( function watch( runId: string, patch: Patch, set: any, save: (body: Record) => Promise, after = 0, + question = '', retry?: (reason: string) => boolean, ) { const stop = streamRun(runId, (event: RunEvent) => { const d: any = event.data ?? {} @@ -510,14 +620,18 @@ function watch( case 'run.finished': // 成果在事件里就有，不必再取一次 run；但 run 对象带着 usage 和 // run_class（出具横幅要用），所以还是拉一次，失败也不影响成果 - patch(() => ({ - phase: 'done', status: PHASE_TEXT.done, output: d.output ?? null, - })) - set({ busy: false, cancel: null }) + // 成果**先落库**再显示。两件事顺序不能并： + // - 落库要马上，否则复核期间关掉标签页，这次跑出来的东西就没了； + // - 显示要等复核，否则用户已经把结论读完、信了，说明才姗姗来迟—— + // 实测复核要十秒上下，这十秒里他看到的是一个没人核对过的答案。 + // 所以这里只改状态，output 交给 settle 在核完之后一次性摆出来 + patch(() => ({ status: '正在核对结果…' })) void save({ status: 'done', answer: answerText(d.output ?? null) }) void api.runs.get(runId) .then((run) => patch(() => ({ run }))) .catch(() => undefined) + // 跑完不等于答对：先接住这次编排，确认没问题再交付 + void settle(runId, question, d.output ?? null, patch, set, save, retry) break } }, undefined, after) @@ -527,4 +641,65 @@ function watch( set({ cancel: stop }) } + +/** + * 接住一次编排的产出，确认能不能就这么交给用户。 + * + * 「跑完了」和「答得对」是两回事。库里能查到大量 status=succeeded 的运行， + * 过程中检索降级了、工具报错了、agent 步数用满了，而交到用户手上的答案对此 + * 只字不提——最糟的一次交出去的干脆是一句内部管道文案。 + * + * 后端先用规则扫一遍事件流，干净的运行直接返回 ok，一次模型都不调；扫出异常 + * 才叫模型重组答案并把话说破（backend/app/engine/review.py）。 + * + * 这个函数必须**在任何情况下都把答案交出去**：复核是补一层说明，它自己失败了 + * 不能把已经到手的结果一起带走，更不能让这一轮永远停在"正在核对"。 + */ +async function settle( + runId: string, question: string, output: Record | null, + patch: Patch, set: any, save: (body: Record) => Promise, + retry?: (reason: string) => boolean, +) { + let result: ReviewResult | null = null + try { + // 复核慢到一定程度就不等了。它是补一层说明，不该反过来把答案扣住不放—— + // 干净的运行后端 15ms 就返回，会走到这个上限的都是真叫了模型的那些 + result = await Promise.race([ + api.copilot.review({ run_id: runId, question }), + new Promise((r) => setTimeout(() => r(null), REVIEW_TIMEOUT_MS)), + ]) + } catch { /* 复核用不了就按原样交付，下面照常走 */ } + + // 库里存完整结论（包括 verdict='ok'），store 里只留值得摆到界面上的那部分。 + // 两者分开，"这一轮没复核过"和"复核过、没发现问题"才是两件可区分的事—— + // 事后排查"它当时为什么没警告我"，靠的就是这个区别 + const review = result && result.verdict !== 'ok' ? result : null + + // 答案根本不可信，而且原因说得清楚——重建一次图再跑，别让用户自己发现 + if (review?.retry && review.severity === 'broken' && retry) { + const reason = review.signals.map((s) => s.detail).join('；') || review.note + if (retry(reason)) return + } + + const rewritten = review?.answer ?? null + patch(() => ({ + phase: 'done', + status: PHASE_TEXT.done, + review, + // 到这一步才把成果摆出来：复核说明和结论同时出现，而不是结论先读完 + output, + // 改写是有损的，原件得留着——界面上可以展开对照。它跟着 review 一起 + // 落库，所以刷新之后还在 + ...(rewritten + ? { rawOutput: { answer: review?.original ?? answerText(output) }, output: { answer: rewritten } } + : {}), + })) + set({ busy: false, cancel: null }) + void save({ + status: 'done', + answer: rewritten || answerText(output), + ...(result ? { review: result } : {}), + }) +} + export { decodeRun } diff --git a/frontend/src/types.ts b/frontend/src/types.ts index e93569d9..781fb61f 100644 --- a/frontend/src/types.ts +++ b/frontend/src/types.ts @@ -222,5 +222,36 @@ export interface ConversationTurn { run_id?: string | null status: 'running' | 'done' | 'error' error: string + /** 复核结论。null = 这一轮没复核过，和「复核过、没发现问题」不是一回事 */ + review?: ReviewResult | null created_at?: string } + +/** + * 一次运行跑完之后的复核结论（后端 engine/review.py）。 + * + * 「跑完了」和「答得对」是两回事：检索降级、工具报错、agent 步数用满这些事 + * 原先一件都不会进到答案里，用户拿到一个看起来很完整的结论，却无从知道它是在 + * 什么条件下得出的。复核层就是把这段补上。 + */ +export interface ReviewSignal { + kind: string + detail: string + /** broken = 答案不可信；degraded = 能用但有缺口 */ + severity: 'broken' | 'degraded' +} + +export interface ReviewResult { + /** ok=没问题 / annotated=加了说明 / rewritten=重写过 / reverted=重写里有新数字被退回 */ + verdict: 'ok' | 'annotated' | 'rewritten' | 'reverted' + /** 给用户看的异常说明。排在成果上方 */ + note: string + /** 重写后的答案。null 表示沿用原答案 */ + answer: string | null + /** 改写之前的那份。改写是有损的，得能对照原件 */ + original?: string | null + /** 换个配置重跑一遍大概率就好了 */ + retry: boolean + severity: 'broken' | 'degraded' | '' + signals: ReviewSignal[] +} diff --git a/scripts/check-stream.mjs b/scripts/check-stream.mjs index db938ea2..63709762 100644 --- a/scripts/check-stream.mjs +++ b/scripts/check-stream.mjs @@ -155,6 +155,42 @@ console.log('\n=== 没查库的那一轮 ===') await page.close() } +console.log('\n=== 复核说明 ===') +{ + // 「跑完了」和「答得对」是两回事。复核说明必须排在成果**上方**——排在下面 + // 等于让人读完整个结论、信了，才知道它是在什么条件下得出的 + const page = await browser.newPage({ viewport: { width: 1280, height: 900 } }) + await page.goto(`${WEB}/preview.html?review=1`, { waitUntil: 'networkidle' }) + await page.waitForTimeout(300) + const body = await page.locator('body').innerText() + + check('说破了取数没跑通', body.includes('取数没跑通')) + check('说明排在结论前面', + body.indexOf('取数没跑通') < body.indexOf('大约 5 个'), + `说明@${body.indexOf('取数没跑通')} 结论@${body.indexOf('大约 5 个')}`) + check('答案本身照常渲染', body.includes('大约 5 个')) + check('异常清单默认收起', !body.includes('no such table')) + check('数得出有几处异常', body.includes('检测到 2 处异常')) + + // broken 用报警色，degraded 用普通边框。混成一个样子，用户就学会了一概忽略 + const colors = await page.evaluate(() => { + const err = getComputedStyle(document.documentElement).getPropertyValue('--err').trim() + const hit = [...document.querySelectorAll('div[style*="border-color"]')] + .map((el) => getComputedStyle(el).borderTopColor) + return { err, hit } + }) + check('两档复核长得不一样', new Set(colors.hit).size > 1, colors.hit.join(' / ')) + + // 改写是有损的，原件得留着能对照 + check('改写前的原文可以展开', body.includes('看改写前的原文')) + check('原文默认不占地方', !body.includes('平台管理员可以修改他人权限。\n')) + + const overflow = await page.evaluate(() => + document.documentElement.scrollWidth - document.documentElement.clientWidth) + check('页面不横向溢出', overflow <= 0, `${overflow}px`) + await page.close() +} + console.log('\n=== 空态 ===') { const page = await browser.newPage({ viewport: { width: 1280, height: 900 } })