diff --git a/backend/app/api/copilot.py b/backend/app/api/copilot.py index 8836c280..92262407 100644 --- a/backend/app/api/copilot.py +++ b/backend/app/api/copilot.py @@ -112,8 +112,10 @@ NODE_REFERENCE = """\ - output：出口，收集最终成果。config.fields = [{name, value}]，value 里写模板引用 - llm：单次模型调用。config: {system, prompt, model, temperature, max_tokens, assign_to, output_schema} - agent：带工具循环的 agent。config: {system, prompt, tools:[工具名], approval, assign_to} - **不要写 max_steps**。平台默认值是按「一步只调一个工具」校准过的；写死一个 - 小数字（见过 8）会让它查完表结构就没额度回答了，用户拿到的是半截结论 + **默认不要写 max_steps**。平台默认值是按「一步只调一个工具」校准过的；写死一个 + 小数字（见过 8）会让它查完表结构就没额度回答了，用户拿到的是半截结论。 + 唯一的例外：上一次就是因为**步数用尽**没跑完（会在改写理由里写明）， + 这时要显式写一个更大的值（比如 24）——那是这种失败唯一的修法 - supervisor：多 agent 协作。config: {goal, agents:[{name, description, system, tools, model}], max_rounds} - tool：直接调一个工具。config: {tool: 工具名, args: {...}, assign_to} - code：沙箱里跑代码。config: {language: python|bash|node, code, timeout, network, assign_to} diff --git a/backend/app/core/config.py b/backend/app/core/config.py index bbfdb269..28bfd46c 100644 --- a/backend/app/core/config.py +++ b/backend/app/core/config.py @@ -65,7 +65,10 @@ class Settings(BaseSettings): # --- 执行引擎护栏 --- max_run_seconds: int = 600 max_graph_steps: int = 200 - max_agent_steps: int = 25 + # 节点上配多少步都过不去这个硬顶。默认值仍是节点上的 12（nodeDefs.ts）， + # 这里只决定"想调大的人最多能调到哪"——实测 87% 的 agent 节点 ≤10 步就收口， + # 绷紧的是那 10%，不该让所有人陪着多花钱 + max_agent_steps: int = 100 #: 上传文件的大小上限（MB）。原来写死 10，而且是读完整个文件才检查—— #: 传一个 1GB 的文件，内存在报 413 之前就吃掉了 max_upload_mb: int = 50 diff --git a/backend/app/engine/nodes/llm.py b/backend/app/engine/nodes/llm.py index eb05ad85..9787b557 100644 --- a/backend/app/engine/nodes/llm.py +++ b/backend/app/engine/nodes/llm.py @@ -353,6 +353,12 @@ async def run_agent(state: GraphState, ctx: NodeContext) -> dict[str, Any]: async def llm_step(step: int, payload: list[BaseMessage]) -> BaseMessage: return await _invoke_streaming(model, payload, ctx, model_id) + @task + async def settle_step(payload: list[BaseMessage]) -> BaseMessage: + # 用 base_model 而不是上面那个绑过工具的 model：收尾轮必须在**结构上** + # 发不出工具调用，靠提示词说"别调工具"是约束不住的 + return await _invoke_streaming(base_model, payload, ctx, model_id) + @task async def tool_step(step: int, name: str, args: dict[str, Any]) -> str: tool = tool_map[name] @@ -479,24 +485,58 @@ async def run_agent(state: GraphState, ctx: NodeContext) -> dict[str, Any]: {"tool": call.get("name", ""), "args": call.get("args", {}) or {}, "skipped": True} ) else: - # 步数用完了。成果只能取模型自己说过的话——messages[-1] 很可能是一条 - # ToolMessage：工具的原始返回，或者串行模式下那句"本轮只执行了第一个 - # 工具"。后者真的漏到用户面前当过答案（run dd9927e6），一句内部管道 - # 文案冒充结论，比明说"没跑完"糟糕得多。 - final_text = next( - (text for m in reversed(messages) - if isinstance(m, AIMessage) and (text := message_text(m).strip())), - "", - ) - hint = ( - f"agent 用满了 {max_steps} 步还没给出结论。" - "把节点上的「最大步数」调大；如果它大部分步数花在逐张表查结构上，" - "也可以在提示词里点明该查哪几张表。" - ) - # 一步一工具时步数消耗得比并行快得多，这里不说清楚，用户只会看到 - # 一个没头没尾的答案，而不知道是被步数掐断的 - final_text = f"{final_text}\n\n（{hint}）".strip() if final_text else f"（{hint}）" - ctx.emit(EventType.LOG, level="warn", message=hint, code="step_limit") + # 步数用完了。走到这儿一定意味着最后那步**要求了工具**（不要工具会 break）， + # 也就是说模型从来没拿到过"说结论"的那一轮——它只是被掐断在半路。 + # + # 所以先补上那一轮：不给工具，让它基于已经查到的东西收口。这比把中间 + # 过程当答案交出去强得多，也比只留一句抱怨强——用户要的是"已知什么、 + # 还缺什么"，不是"系统哪里不够用"。 + settled = "" + messages.append(HumanMessage(content=( + f"步数预算用完了（{max_steps} 步），现在起不能再调用任何工具。" + "基于已经查到的信息给出结论；明确说清哪些部分没有查到、" + "结论因此有什么局限。不要编造没查到的数据。" + ))) + try: + response = await settle_step(messages) + usage = _usage_of(response, model_id) + for key in ("input_tokens", "output_tokens", "calls"): + total_usage[key] += usage[key] + total_usage["cost_usd"] = round(total_usage["cost_usd"] + usage["cost_usd"], 6) + settled = message_text(response).strip() + except Exception as e: # noqa: BLE001 - 收尾失败不能把已有成果一起赔进去 + ctx.emit(EventType.LOG, level="warn", + message=f"收尾轮没跑成：{type(e).__name__}: {e}", code="settle_failed") + + if settled: + final_text = settled + hint = ( + f"agent 用满了 {max_steps} 步，这个结论是基于已经查到的部分给出的。" + "要跑全请把节点上的「最大步数」调大；如果它大部分步数花在逐张表" + "查结构上，也可以在提示词里点明该查哪几张表。" + ) + # 和 step_limit 分开发：库里那些老事件的含义确实是"硬截断"， + # 复用同一个 code 会把历史运行重新解释成另一回事 + ctx.emit(EventType.LOG, level="warn", message=hint, code="step_limit_settled") + else: + # 收尾轮也没说出话。成果只能取模型自己说过的话——messages[-1] 很可能 + # 是一条 ToolMessage：工具的原始返回，或者串行模式下那句"本轮只执行了 + # 第一个工具"。后者真的漏到用户面前当过答案（run dd9927e6），一句内部 + # 管道文案冒充结论，比明说"没跑完"糟糕得多。 + final_text = next( + (text for m in reversed(messages) + if isinstance(m, AIMessage) and (text := message_text(m).strip())), + "", + ) + hint = ( + f"agent 用满了 {max_steps} 步还没给出结论。" + "把节点上的「最大步数」调大；如果它大部分步数花在逐张表查结构上，" + "也可以在提示词里点明该查哪几张表。" + ) + # 一步一工具时步数消耗得比并行快得多，这里不说清楚，用户只会看到 + # 一个没头没尾的答案，而不知道是被步数掐断的 + final_text = f"{final_text}\n\n（{hint}）".strip() if final_text else f"（{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/review.py b/backend/app/engine/review.py index b28f23fc..0d318bc1 100644 --- a/backend/app/engine/review.py +++ b/backend/app/engine/review.py @@ -58,6 +58,12 @@ WARN_CODES: dict[str, tuple[str, str]] = { "step_limit": ("step_limit", BROKEN), "empty_completion": ("empty_completion", BROKEN), # 能用但有缺口 + # 步数用尽，但收尾轮让模型基于已查到的部分收了口：答案是完整的一段话， + # 只是覆盖面不全——这是缺口不是断裂，所以不判 BROKEN。 + # 连带效果是不再触发自动重跑，那正是想要的：同样配置再跑一遍还撞同一堵墙 + "step_limit_settled": ("step_limit_settled", DEGRADED), + # 收尾轮自己挂了。影响由同时发出的 step_limit(BROKEN) 承担，这条是说清原因 + "settle_failed": ("settle_failed", DEGRADED), "loop_limit": ("loop_limit", DEGRADED), "tool_args_fixed": ("tool_args_fixed", DEGRADED), "node_retry": ("node_retry", DEGRADED), diff --git a/backend/tests/test_agent_steps.py b/backend/tests/test_agent_steps.py new file mode 100644 index 00000000..45bddb6f --- /dev/null +++ b/backend/tests/test_agent_steps.py @@ -0,0 +1,218 @@ +"""步数用尽时，agent 要收口，而不是把半截过程交出去。 + +走到步数上限一定意味着最后那步**要求了工具**（不要工具的话循环早 break 了）， +也就是说模型从来没拿到过"说结论"的那一轮——它只是被掐断在半路。交出去的 +是它上一句中间过程的话，末尾补一句"步数不够"。 + +所以补上那一轮：不给工具，让它基于已经查到的收口。这里守三件事——收尾轮 +真的发生了、它**发不出**工具调用、以及它没有变成每次都多烧的一次调用。 + +顺带补回 agent 循环的集成测试：原来那个文件在脱敏那轮被整个删掉了。 +""" + +from __future__ import annotations + +import asyncio + +import pytest +from langchain_core.messages import AIMessage +from sqlalchemy import select + +from app.db.base import SessionLocal +from app.db.models import Run, RunEvent +from app.engine.runner import run_manager + +#: 收尾轮该说的话。挑一句和"中间过程"截然不同的，好分辨答案到底取自哪一轮 +SETTLED = "已经查到的部分：门店 8 家。其余未查到，这个数是下限。" +MIDWAY = "中间过程：还得再查一张表。" + +MAX_STEPS = 3 + + +@pytest.fixture +async def runner(): + await run_manager.setup() + yield run_manager + await run_manager.shutdown() + + +def _graph(*, max_steps: int = MAX_STEPS) -> dict: + return { + "nodes": [ + {"id": "in", "type": "input", "data": {"config": {}}}, + {"id": "work", "type": "agent", "data": {"config": { + "prompt": "{{ input.question }}", + "tools": ["calculator"], + "max_steps": max_steps, + "approval": "never", + }}}, + {"id": "out", "type": "output", "data": {"config": {}}}, + ], + "edges": [{"source": "in", "target": "work"}, {"source": "work", "target": "out"}], + } + + +def never_concludes(monkeypatch, *, settle_raises: bool = False) -> None: + """绑了工具就永远再要一次工具；没绑工具才说话。 + + `self.tools` 是 bind_tools 之后才有的。收尾轮走的是**没绑过工具**的 + base_model，所以它必然落到下面那个分支——这正是"收尾轮发不出工具调用" + 那条断言的着力点：真要是绑着工具过来的，拿到的就会是 MIDWAY。 + """ + from app.providers import mock_model + + def _decide(self, messages): + if self.tools: + return AIMessage( + content=MIDWAY, + tool_calls=[{"name": self.tools[0]["name"], + "args": {"expression": "1+1"}, + "id": f"call_{len(messages)}"}], + ) + if settle_raises: + raise RuntimeError("收尾轮炸了") + return AIMessage(content=SETTLED) + + monkeypatch.setattr(mock_model.MockChatModel, "_decide", _decide) + + +async def _run(graph: dict, question: str = "有几家门店"): + run = await run_manager.start(graph=graph, input_payload={"question": question}) + for _ in range(200): + await asyncio.sleep(0.05) + async with SessionLocal() as session: + row = (await session.execute(select(Run).where(Run.id == run.id))).scalar_one() + if row.status in ("succeeded", "failed", "cancelled"): + break + else: + pytest.fail("运行没有在 10 秒内结束") + async with SessionLocal() as session: + events = list((await session.execute( + select(RunEvent).where(RunEvent.run_id == run.id).order_by(RunEvent.seq) + )).scalars()) + return events, row + + +def codes(events) -> list[str]: + return [e.data.get("code") for e in events + if e.type == "log" and (e.data or {}).get("level") == "warn"] + + +def answer_of(row) -> str: + from app.engine import review as rv + return rv.answer_text(row.output) + + +# -------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_step_limit_ends_with_a_conclusion_not_a_stump(runner, monkeypatch) -> None: + """这块改动的全部意义：交出去的是结论，不是中间过程加一句抱怨。""" + never_concludes(monkeypatch) + events, row = await _run(_graph()) + assert row.status == "succeeded", row.error + + text = answer_of(row) + assert SETTLED in text, text + assert MIDWAY not in text, f"上一句中间过程冒充了答案：{text}" + assert "step_limit_settled" in codes(events), codes(events) + + +@pytest.mark.asyncio +async def test_the_settle_round_has_no_tools_bound(runner, monkeypatch) -> None: + """收尾轮必须在**结构上**发不出工具调用，而不是靠提示词请它别调。 + + mock 的判据就是"有没有绑工具"：真要是把绑过工具的 model 传进了收尾轮， + 它会照常再要一次工具，答案里出现的就是 MIDWAY。 + """ + never_concludes(monkeypatch) + events, row = await _run(_graph()) + + assert SETTLED in answer_of(row) + # 工具调用次数正好等于步数——收尾轮一次都没多调 + assert len([e for e in events if e.type == "tool.start"]) == MAX_STEPS + + +@pytest.mark.asyncio +async def test_the_settle_round_costs_exactly_one_extra_call(runner, monkeypatch) -> None: + """多花的必须是可数的一次，不是一个说不清的尾巴。""" + never_concludes(monkeypatch) + events, _ = await _run(_graph()) + assert len([e for e in events if e.type == "llm.start"]) == MAX_STEPS + 1 + + +@pytest.mark.asyncio +async def test_a_normal_agent_does_not_pay_for_a_settle_round(runner, monkeypatch) -> None: + """反向断言：正常收口的 agent 一分钱都不该多花。 + + 没有这条，"每次都多跑一轮"这种退化会悄无声息地通过全部其他测试。 + """ + from app.providers import mock_model + + monkeypatch.setattr(mock_model.MockChatModel, "_decide", + lambda self, messages: AIMessage(content="直接就有结论了。")) + events, row = await _run(_graph()) + + assert "直接就有结论了。" in answer_of(row) + assert len([e for e in events if e.type == "llm.start"]) == 1 + assert not [c for c in codes(events) if c and c.startswith("step_limit")] + + +@pytest.mark.asyncio +async def test_a_failed_settle_round_falls_back_and_keeps_what_was_found( + runner, monkeypatch +) -> None: + """收尾轮自己挂了，不能把已经查到的一起赔进去。""" + never_concludes(monkeypatch, settle_raises=True) + events, row = await _run(_graph()) + assert row.status == "succeeded", row.error + + text = answer_of(row) + assert "settle_failed" in codes(events), codes(events) + # 退回原来的行为：老的 step_limit（BROKEN），而不是 settled 那个 + assert "step_limit" in codes(events) + assert "step_limit_settled" not in codes(events) + # 已有成果还在，末尾挂着说明 + assert MIDWAY in text and "用满了" in text, text + + +def test_max_steps_is_clamped_by_the_hard_cap() -> None: + """节点上写多大都过不去硬顶——否则一张图就能把预算烧穿。""" + from app.core.config import settings + + assert min(200, settings.max_agent_steps) == settings.max_agent_steps + assert settings.max_agent_steps >= 25, "抬高上限不该反而把原来能配的值挡回去" + + +# -------------------------------------------------------------------------- +# 复核层怎么看这两种收场 +# -------------------------------------------------------------------------- + + +def _warn(message: str, code: str) -> dict: + return {"type": "log", "node_id": "work", + "data": {"level": "warn", "message": message, "code": code}} + + +OK_OUTPUT = {"text": SETTLED} + + +def test_a_settled_run_is_a_gap_not_a_break() -> None: + """收口过的答案是完整一段话，只是覆盖面不全——那是缺口不是断裂。""" + from app.engine import review as rv + + signals = rv.scan([_warn("用满了 3 步，基于已查到的部分给出", "step_limit_settled")], + OK_OUTPUT) + assert [s.severity for s in signals] == [rv.DEGRADED] + # 连带效果正是想要的：不再触发自动重跑 + assert not rv.should_retry(signals) + + +def test_a_hard_truncation_is_still_broken_and_still_retries() -> None: + """收尾轮没能收上来，判定一点没变——这条是防止改动把老路一起改坏。""" + from app.engine import review as rv + + signals = rv.scan([_warn("用满了 3 步还没给出结论", "step_limit")], OK_OUTPUT) + assert rv.worst(signals) == rv.BROKEN + assert rv.should_retry(signals) diff --git a/frontend/src/canvas/nodeDefs.ts b/frontend/src/canvas/nodeDefs.ts index 70510e81..374d9393 100644 --- a/frontend/src/canvas/nodeDefs.ts +++ b/frontend/src/canvas/nodeDefs.ts @@ -133,7 +133,7 @@ export const NODE_DEFS: Record = { { key: 'prompt', label: '任务', type: 'prompt' }, { key: 'tools', label: '可用工具', type: 'tools' }, { key: 'skills', label: '挂载 Skill', type: 'skills' }, - { key: 'max_steps', label: '最大步数', type: 'number', min: 1, max: 25 }, + { key: 'max_steps', label: '最大步数', type: 'number', min: 1, max: 100 }, { key: 'approval', label: '工具审批', type: 'select', options: [