diff --git a/backend/app/api/copilot.py b/backend/app/api/copilot.py index 538bbf4e..2d41a96c 100644 --- a/backend/app/api/copilot.py +++ b/backend/app/api/copilot.py @@ -111,7 +111,9 @@ NODE_REFERENCE = """\ - input：入口。config.fields = [{name, required, default, description}] - output：出口，收集最终成果。config.fields = [{name, value}]，value 里写模板引用 - llm：单次模型调用。config: {system, prompt, model, temperature, max_tokens, assign_to, output_schema} -- agent：带工具循环的 agent。config: {system, prompt, tools:[工具名], max_steps, approval, assign_to} +- agent：带工具循环的 agent。config: {system, prompt, tools:[工具名], approval, assign_to} + **不要写 max_steps**。平台默认值是按「一步只调一个工具」校准过的；写死一个 + 小数字（见过 8）会让它查完表结构就没额度回答了，用户拿到的是半截结论 - 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/engine/nodes/llm.py b/backend/app/engine/nodes/llm.py index e96d0a43..3bf5fb83 100644 --- a/backend/app/engine/nodes/llm.py +++ b/backend/app/engine/nodes/llm.py @@ -17,8 +17,21 @@ from app.db.models import Skill from app.engine.context import NodeContext, NodeError from app.engine.state import GraphState, message_text, template_context, thinking_text from app.providers import catalog -from app.providers.factory import ModelSpec, ProviderNotConfigured, get_chat_model -from app.tools.registry import ToolContext, build_tools, get_spec +from app.providers.factory import ( + ModelSpec, + ProviderNotConfigured, + bind_tools_safely, + get_chat_model, +) +from app.tools.registry import ( + ToolArgsError, + ToolContext, + args_model_of, + build_tools, + describe_args, + get_spec, + prepare_args, +) # -------------------------------------------------------------------------- # 公共部分 @@ -270,6 +283,33 @@ async def _resolve_tools(ctx: NodeContext, state: GraphState) -> list[BaseTool]: return await build_tools(names, tool_ctx, session=session) +SKIPPED_NOTE = ( + "本轮只执行了第一个工具。看到它的结果之后，再决定这一个还要不要调、要用什么参数调。" +) + + +def split_tool_calls( + tool_calls: list[dict[str, Any]], *, parallel: bool +) -> tuple[list[dict[str, Any]], list[tuple[dict[str, Any], ToolMessage]]]: + """把一轮里的工具调用分成"这次跑"和"这次不跑"。 + + 串行时只跑第一个。剩下的不能就这么丢掉：两家 API 都要求每个 tool_use 在 + 下一轮有一条配对的 tool_result，少一条下次调用直接 400。所以这里给每个 + 未执行的调用配一条说明性的 ToolMessage 一起交出去。 + + 它们也**绝不能**发 tool.start —— 前端靠 call_id 配对 start/end + （frontend/src/run/decode.ts），发了一条没有结尾的 start，那个工具卡片就会 + 在界面上永远转圈。所以这个函数不碰事件，只产消息。 + """ + if parallel or len(tool_calls) <= 1: + return list(tool_calls), [] + deferred = [ + (call, ToolMessage(content=SKIPPED_NOTE, tool_call_id=call.get("id") or f"skipped-{i}")) + for i, call in enumerate(tool_calls[1:]) + ] + return tool_calls[:1], deferred + + def _needs_approval(ctx: NodeContext, tool_name: str) -> bool: mode = ctx.cfg("approval", "dangerous") # never | dangerous | always if mode == "always": @@ -295,10 +335,13 @@ async def run_agent(state: GraphState, ctx: NodeContext) -> dict[str, Any]: tools = await _resolve_tools(ctx, state) tool_map = {t.name: t for t in tools} - model = base_model.bind_tools(tools) if tools else base_model + # 默认串行：一轮只调一个工具，拿到结果再想下一步。并行更快，但一批工具 + # 中间没有任何新的思考——第二个调用是照着"还没看到第一个结果"时的判断发出来的 + parallel_tools = bool(ctx.cfg("parallel_tools", False)) + model = bind_tools_safely(base_model, tools, parallel=parallel_tools) messages = await _build_messages(state, ctx) - max_steps = min(int(ctx.cfg("max_steps", 8) or 8), settings.max_agent_steps) + max_steps = min(int(ctx.cfg("max_steps", 12) or 12), settings.max_agent_steps) total_usage: dict[str, Any] = {"input_tokens": 0, "output_tokens": 0, "cost_usd": 0.0, "calls": 0} transcript: list[dict[str, Any]] = [] @@ -328,7 +371,9 @@ async def run_agent(state: GraphState, ctx: NodeContext) -> dict[str, Any]: final_text = message_text(response) break - for call in tool_calls: + run_calls, deferred = split_tool_calls(tool_calls, parallel=parallel_tools) + + for call in run_calls: name = call.get("name", "") args = call.get("args", {}) or {} call_id = call.get("id") or f"{ctx.node.id}-{step}-{name}" @@ -339,6 +384,24 @@ async def run_agent(state: GraphState, ctx: NodeContext) -> dict[str, Any]: ) continue + # 参数先过 schema 再决定要不要打扰人审批：参数就不对的调用没必要问人， + # 而纠正过的参数必须是人在审批框里看到的那一份 + schema = args_model_of(tool_map[name]) + if schema is not None: + try: + args, fix_note = prepare_args(schema, args) + except ToolArgsError as e: + # 不执行，把"它接受什么"原样喂回去。模型下一步照着改就行， + # 这才是失败之后真正能自愈的重试——以前喂回去的是一句 + # TypeError，里面既没有正确参数名也没有字段说明 + ctx.emit(EventType.TOOL_ERROR, tool=name, call_id=call_id, error=str(e)) + messages.append(ToolMessage(content=str(e), tool_call_id=call_id)) + transcript.append({"tool": name, "args": args, "ok": False, "result": str(e)}) + continue + if fix_note: + ctx.emit(EventType.LOG, level="warn", + message=f"工具 {name}：{fix_note}") + if _needs_approval(ctx, name): ctx.emit( EventType.HUMAN_REQUESTED, @@ -377,6 +440,10 @@ async def run_agent(state: GraphState, ctx: NodeContext) -> dict[str, Any]: ok = True except Exception as e: # noqa: BLE001 - 工具失败要喂回模型，让它自己纠错 content = f"工具执行失败：{type(e).__name__}: {e}" + if isinstance(e, TypeError) and schema is not None: + # 签名对不上还能走到这儿，说明 schema 和函数本身不一致（工具的 bug）。 + # 模型改不了这个，但把参数表摊开至少让它别在同一个地方反复试 + content += f"\n{describe_args(schema, args)}" ok = False elapsed = int((time.perf_counter() - started) * 1000) snapshot_id: str | None = None @@ -402,13 +469,32 @@ async def run_agent(state: GraphState, ctx: NodeContext) -> dict[str, Any]: {"tool": name, "args": args, "ok": ok, "duration_ms": elapsed, "result": content[:4000]} ) messages.append(ToolMessage(content=content, tool_call_id=call_id)) + + # 没跑的那些排在执行过的后面，保持模型原本的调用顺序 + for call, message in deferred: + messages.append(message) + transcript.append( + {"tool": call.get("name", ""), "args": call.get("args", {}) or {}, "skipped": True} + ) else: - final_text = message_text(messages[-1]) if messages else "" - ctx.emit( - EventType.LOG, - level="warn", - message=f"agent 达到最大步数 {max_steps}，提前收尾", + # 步数用完了。成果只能取模型自己说过的话——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) 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 18390e35..5cf702d6 100644 --- a/backend/app/engine/nodes/multi.py +++ b/backend/app/engine/nodes/multi.py @@ -14,8 +14,14 @@ from app.db.models import Workflow from app.engine.context import NodeContext, NodeError from app.engine.state import GraphState, message_text, template_context from app.providers import catalog -from app.providers.factory import ModelSpec, get_chat_model -from app.tools.registry import ToolContext, build_tools +from app.providers.factory import ModelSpec, bind_tools_safely, get_chat_model +from app.tools.registry import ( + ToolArgsError, + ToolContext, + args_model_of, + build_tools, + prepare_args, +) _MAX_DEPTH = 3 @@ -122,7 +128,7 @@ async def run_supervisor(state: GraphState, ctx: NodeContext) -> dict[str, Any]: ) tools = await build_tools(cfg.get("tools", []) or [], tool_ctx, session=session) - bound = model.bind_tools(tools) if tools else model + bound = bind_tools_safely(model, tools, parallel=False) tool_map = {t.name: t for t in tools} messages: list[Any] = [ SystemMessage(content=cfg.get("system") or f"你是 {name}。"), @@ -140,15 +146,32 @@ async def run_supervisor(state: GraphState, ctx: NodeContext) -> dict[str, Any]: tname = call.get("name", "") targs = call.get("args", {}) or {} cid = call.get("id") or f"{name}-{tname}" - ctx.emit(EventType.TOOL_START, tool=tname, args=targs, agent=name, call_id=cid) if tname not in tool_map: + ctx.emit(EventType.TOOL_START, tool=tname, args=targs, agent=name, call_id=cid) content = f"错误：没有名为 {tname} 的工具" else: - try: - raw = await tool_map[tname].ainvoke(targs) - content = raw if isinstance(raw, str) else json.dumps(raw, ensure_ascii=False, default=str) - except Exception as e: # noqa: BLE001 - content = f"工具失败：{type(e).__name__}: {e}" + schema = args_model_of(tool_map[tname]) + fix_note = None + args_error = None + if schema is not None: + try: + targs, fix_note = prepare_args(schema, targs) + except ToolArgsError as e: + args_error = str(e) + # 纠正后的参数才是真正要执行的那份，tool.start 要发它 + 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}") + if args_error: + # 参数就不对，没必要真调一次。把"它接受什么"喂回去让它改 + content = args_error + else: + try: + raw = await tool_map[tname].ainvoke(targs) + content = raw if isinstance(raw, str) else json.dumps(raw, ensure_ascii=False, default=str) + except Exception as e: # noqa: BLE001 + content = f"工具失败：{type(e).__name__}: {e}" ctx.emit(EventType.TOOL_END, tool=tname, agent=name, call_id=cid, preview=content[:1500]) calls.append({"tool": tname, "args": targs, "result": content[:2000]}) messages.append(ToolMessage(content=content, tool_call_id=cid)) diff --git a/backend/app/engine/nodes/tools.py b/backend/app/engine/nodes/tools.py index 09ad34ee..e0f062f9 100644 --- a/backend/app/engine/nodes/tools.py +++ b/backend/app/engine/nodes/tools.py @@ -12,7 +12,7 @@ from app.engine.context import NodeContext, NodeError from app.engine.state import GraphState from app.sandbox.base import SandboxLimits from app.sandbox.manager import sandbox_manager -from app.tools.registry import ToolContext, call_tool, get_spec +from app.tools.registry import ToolArgsError, ToolContext, call_tool, get_spec def _tool_ctx(ctx: NodeContext) -> ToolContext: @@ -55,11 +55,25 @@ async def run_tool(state: GraphState, ctx: NodeContext) -> dict[str, Any]: ctx.emit(EventType.TOOL_START, tool=name, args=args) started = time.perf_counter() + + def _fixed(note: str) -> None: + # 替它跑通了这一次，但节点配置里那个错的参数名原封不动，下次还会踩。 + # 所以纠正必须留一条看得见的痕迹，而不是安静地把事办了 + ctx.emit(EventType.LOG, level="warn", + message=f"工具 {name}：{note}。请到节点里改正。") + try: async with SessionLocal() as session: - result = await call_tool(name, args, _tool_ctx(ctx), session=session) + result = await call_tool(name, args, _tool_ctx(ctx), session=session, on_fix=_fixed) except KeyError as e: raise NodeError(ctx.node.id, str(e)) from e + except ToolArgsError as e: + # 参数对不上且没有唯一候选可纠。报错里已经写清楚该填什么， + # 不要再套一层 "执行失败：" 把它推远 + ctx.emit(EventType.TOOL_ERROR, tool=name, error=str(e)) + if ctx.cfg("fail_fast", True): + raise NodeError(ctx.node.id, f"工具 {name}：{e}") from e + result = {"error": str(e)} except Exception as e: # noqa: BLE001 ctx.emit(EventType.TOOL_ERROR, tool=name, error=f"{type(e).__name__}: {e}") if ctx.cfg("fail_fast", True): diff --git a/backend/app/providers/catalog.py b/backend/app/providers/catalog.py index e5ac99a0..4bdd4558 100644 --- a/backend/app/providers/catalog.py +++ b/backend/app/providers/catalog.py @@ -167,6 +167,25 @@ _ADAPTIVE_THINKING_PREFIXES = _NO_SAMPLING_PREFIXES # 支持 output_config.effort _EFFORT_PREFIXES = _NO_SAMPLING_PREFIXES + ("claude-opus-4-5",) +# 交错思考（工具结果回来之后再想一段，而不是一轮开头想完就只剩执行）。 +# +# 4.6 及以后走 adaptive thinking，交错是自带的，不需要也不该再挂 beta header +# ——在 Opus 4.6 上手动模式压根没有交错思考，只有 adaptive 有。上面那张 +# _ADAPTIVE_THINKING_PREFIXES 覆盖的都属于这一档。 +# +# 需要显式开的是 Claude 4 / 4.5 这一代。目录里没有列它们，但模型名是放行的 +# （用户可以手填），所以这条路径要留着。Haiku 4.5 不支持，前缀也就不包含它。 +_INTERLEAVED_BETA_PREFIXES = ("claude-opus-4", "claude-sonnet-4") +INTERLEAVED_BETA = "interleaved-thinking-2025-05-14" + + +def needs_interleaved_beta(model: str) -> bool: + """这个模型要不要靠 beta header 才会在工具之间思考。""" + name = _norm(model) + if supports_adaptive_thinking(name): + return False # adaptive 自带，多挂一个 header 只会是噪音 + return name.startswith(_INTERLEAVED_BETA_PREFIXES) + def _norm(model: str) -> str: """去掉 openrouter 之类的前缀，只看模型名本身。""" diff --git a/backend/app/providers/factory.py b/backend/app/providers/factory.py index dbfda007..43d67e61 100644 --- a/backend/app/providers/factory.py +++ b/backend/app/providers/factory.py @@ -138,17 +138,49 @@ def build_chat_model(provider: Provider, spec: ModelSpec) -> BaseChatModel: # 会出现"思考完就没额度写答案"，返回一条空正文。 kwargs.setdefault("max_tokens", 8192) + # 交错思考 —— 工具结果回来之后模型还能再想一段，而不是一轮开头想完 + # 就只剩埋头执行。串行跑工具时这一条最要紧：每一步之间本来就该重新 + # 判断一次"看到这个结果之后下一步该干嘛"。 + # + # 两代模型的开法完全不同，这也是为什么这里要分叉： + # 4.6 起 adaptive thinking **自带**交错，不需要 beta header。 + # （Opus 4.6 上手动模式压根没有交错，只有 adaptive 有， + # 所以这一档绝不能退回手动。） + # 4 / 4.5 手动开 thinking + 挂 interleaved-thinking beta header。 + # 目录里没列这一代，但模型名是放行的，用户可以手填。 model_kwargs: dict[str, Any] = {} + thinking_on = False + mode = spec.thinking or "summarized" + if catalog.supports_adaptive_thinking(model): - mode = spec.thinking or "summarized" if mode in ("adaptive", "summarized"): # summarized 会把思考摘要一起返回，画布上就能展示"它在想什么" kwargs["thinking"] = { "type": "adaptive", "display": "summarized" if mode == "summarized" else "omitted", } + thinking_on = True elif mode == "off": kwargs["thinking"] = {"type": "disabled"} + elif mode != "off" and catalog.needs_interleaved_beta(model): + # 光挂 header 是不够的：这一代默认根本没开 thinking， + # 没有思考可交错，header 就是一条什么都不做的空设置 + kwargs["thinking"] = { + "type": "enabled", + "budget_tokens": max(1024, int(kwargs["max_tokens"]) // 2), + } + thinking_on = True + # betas 会把调用改道 client.beta.messages。自建网关可能不认它， + # 而那是请求时才炸的，所以留一个关得掉的口子 + if extra.get("interleaved_thinking") is not False: + kwargs["betas"] = [catalog.INTERLEAVED_BETA] + + # 开了 thinking 就不能再传 temperature（要求是 1）。4.6 起的模型压根 + # 不收采样参数，上面 supports_sampling 已经挡住了；漏的是 4/4.5 这一代 + # ——它们本来收 temperature，一旦开了 thinking 再传就是 400 + if thinking_on: + kwargs.pop("temperature", None) + if spec.effort and catalog.supports_effort(model): model_kwargs["output_config"] = {"effort": spec.effort} if model_kwargs: @@ -228,6 +260,30 @@ class _LazyReasoningChatOpenAI: _ReasoningAwareChatOpenAI = _LazyReasoningChatOpenAI() +def bind_tools_safely(model: BaseChatModel, tools: list[Any], *, parallel: bool) -> Any: + """绑定工具，并在需要时让模型一次只发一个工具调用。 + + `parallel_tool_calls=False` 两家都认，只是落点不同：ChatAnthropic 把它翻成 + `tool_choice={"type": "auto", "disable_parallel_tool_use": True}`（type 是 auto + 而不是强制某个工具，所以和 thinking 并存没问题），ChatOpenAI 原样透传。 + + 但 provider 可以是任意 OpenAI 兼容网关（Ollama、vLLM、OpenRouter…），个别 + 网关可能在**请求时**才拒绝这个字段——那种情况这里的 try 拦不住。真撞上了， + 把节点上的"并行执行工具"打开即可，绑定就退回原样。 + + 无论如何这都只是省 token 的优化：模型仍然多发了几个调用时，编排层一轮只执行 + 第一个，串行语义不依赖这里能不能传成功。 + """ + if not tools: + return model + if parallel: + return model.bind_tools(tools) + try: + return model.bind_tools(tools, parallel_tool_calls=False) + except (TypeError, ValueError, NotImplementedError): + return model.bind_tools(tools) + + async def get_chat_model(session: AsyncSession, spec: ModelSpec) -> tuple[BaseChatModel, str]: """返回 (模型, 实际使用的模型 id)。找不到任何 provider 时回退到 mock， 保证一张图在没配 key 的机器上也能跑完。""" diff --git a/backend/app/tools/datasource.py b/backend/app/tools/datasource.py index e138ac39..3dcc63fc 100644 --- a/backend/app/tools/datasource.py +++ b/backend/app/tools/datasource.py @@ -35,7 +35,12 @@ class _QueryArgs(BaseModel): class _SchemaArgs(BaseModel): - table: str | None = Field(default=None, description="表名；留空则列出所有表") + # 一次能问好几张。串行跑工具时一次调用就是一步，逐张问 6 张表就是 6 步—— + # 实测一次 8 步的运行里 7 步花在这上面，最后只剩一步来真查数据（run dd9927e6） + table: str | list[str] | None = Field( + default=None, + description='表名。可以一次给多张："user" 或 ["user","roles"]；留空则列出所有表', + ) def tool_names(source: DataSource) -> list[str]: @@ -125,9 +130,13 @@ def _make_query_tool(source: DataSource, ctx: ToolContext) -> StructuredTool: def _make_schema_tool(source: DataSource) -> StructuredTool: - async def _run(table: str | None = None) -> str: - if table: - return introspect.describe_table(source, table) + async def _run(table: str | list[str] | None = None) -> str: + wanted = [table] if isinstance(table, str) else list(table or []) + # 逗号分隔也认："user, roles" 和 ["user","roles"] 是同一个意思， + # 而模型两种都会写。为此拒绝一次调用，纯属浪费一步 + wanted = [t.strip() for one in wanted for t in str(one).split(",") if t.strip()] + if wanted: + return "\n\n".join(introspect.describe_table(source, t) for t in wanted) tables = introspect.table_names(source) if not tables: return ( @@ -142,7 +151,8 @@ def _make_schema_tool(source: DataSource) -> StructuredTool: name=f"{SCHEMA_PREFIX}{source.name}", description=( f"查看数据源「{source.name}」的表结构。" - "传 table 看某张表的字段；不传则列出所有表。写 SQL 前先用它确认字段名。" + "写 SQL 前先用它确认字段名。不传 table 则列出所有表；" + "**要看多张表就一次全传进来**（table 可以是数组），不要一张一张问。" ), args_schema=_SchemaArgs, coroutine=_run, diff --git a/backend/app/tools/registry.py b/backend/app/tools/registry.py index 7d506bf2..040e006c 100644 --- a/backend/app/tools/registry.py +++ b/backend/app/tools/registry.py @@ -6,7 +6,7 @@ from dataclasses import dataclass, field from typing import Any, Awaitable, Callable from langchain_core.tools import BaseTool, StructuredTool -from pydantic import BaseModel +from pydantic import BaseModel, ValidationError ToolFunc = Callable[..., Awaitable[Any]] @@ -151,12 +151,157 @@ async def build_tools( return tools -async def call_tool(name: str, args: dict[str, Any], ctx: ToolContext, session: Any = None) -> Any: - """直接调用一个工具（工具节点、以及设置页的"试一下"按钮用）。""" +class ToolArgsError(ValueError): + """工具参数对不上 schema。 + + message 是直接给人和模型看的：报错里必须说清楚"收到了什么"和"它接受什么"， + 因为看到这句话的下一步动作就是照着改参数名。不要再往外包一层。 + """ + + +_TYPE_LABELS = { + "string": "字符串", "integer": "整数", "number": "数字", + "boolean": "布尔", "array": "数组", "object": "对象", +} + + +def _type_label(prop: dict[str, Any]) -> str: + """JSON Schema 的类型写成人话。可选字段是 anyOf[T, null]，要摊开看。""" + names: list[str] = [] + for node in [prop, *(prop.get("anyOf") or [])]: + kind = node.get("type") + if isinstance(kind, str) and kind != "null": + names.append(_TYPE_LABELS.get(kind, kind)) + return "/".join(dict.fromkeys(names)) or "任意" + + +def args_model_of(tool: BaseTool) -> type[BaseModel] | None: + """工具的参数模型。 + + MCP 这类动态工具的 args_schema 可能直接是一份 JSON Schema dict，拿不到 + pydantic 模型——那种只能原样放行，不能因为校验不了就拒绝调用。 + """ + schema = getattr(tool, "args_schema", None) + if isinstance(schema, type) and issubclass(schema, BaseModel): + return schema + return None + + +def describe_args( + schema: type[BaseModel], + got: dict[str, Any] | None = None, + error: ValidationError | None = None, +) -> str: + """把"哪里不对、它到底要什么"写成一句能照着改的话。 + + 字段说明取 Field(description=...)，各工具定义里本来就写了，这里只是搬到报错里 + ——模型拿到这句话就能在下一步自己改对，不用人工介入。 + + 返回的永远是一句完整的话（"参数不对：…。它接受：…"），调用方直接拼在 + 工具名后面即可，不用猜这次会不会缺个主语。 + """ + js = schema.model_json_schema() + props: dict[str, Any] = js.get("properties") or {} + required = [k for k in (js.get("required") or []) if k in props] + + fields: list[str] = [] + for key, prop in props.items(): + tag = "必填" if key in required else "可选" + desc = str(prop.get("description") or "").strip() + fields.append(f"{key}（{tag}，{_type_label(prop)}）" + (f" — {desc}" if desc else "")) + tail = "它接受：" + "；".join(fields) if fields else "它不接受任何参数。" + + got = dict(got or {}) + unknown = [k for k in got if k not in props] + missing = [k for k in required if k not in got] + problems: list[str] = [] + if unknown: + problems.append(f"{'、'.join(unknown)} 不是它的参数") + if missing: + problems.append(f"缺少必填参数 {'、'.join(missing)}") + for err in error.errors() if error else (): + loc = ".".join(str(p) for p in (err.get("loc") or ())) + if not loc or loc in unknown or loc in missing: + continue # 上面两条已经说过了，不重复 + # 用 schema 里的类型说，不要把 pydantic 那句英文原样甩出去 + problems.append( + f"{loc} 的值不合法，应为{_type_label(props[loc])}" + if loc in props + else f"{loc} 的值不合法（{err.get('msg', '')}）" + ) + + head = "参数不对：" + "；".join(problems) + "。" if problems else "参数不对。" + return f"{head}{tail}" + + +def prepare_args( + schema: type[BaseModel], args: dict[str, Any] +) -> tuple[dict[str, Any], str | None]: + """校验工具参数，必要时纠正一个明显写错的参数名。 + + 返回 (可直接展开给函数的 payload, 纠正说明或 None)。纠正说明不为空时调用方 + **必须**把它说出去——参数还错在配置里/还错在模型脑子里，替它跑通一次却不吭声， + 下一次照样踩。 + + 多余的参数名在这里是硬错误，哪怕校验能过：pydantic 默认把认不出的键悄悄丢掉， + 于是全可选字段的 schema 上 {"tabel": "orders"} 会"成功"执行一次什么都没查的调用。 + 这种静默失败比报错难查得多。 + + 纠正只在候选唯一时发生。多于一个就有排列组合，猜对了也是运气，不如把 schema + 摊开让人或模型自己改。 + """ + args = dict(args or {}) + js = schema.model_json_schema() + props: dict[str, Any] = js.get("properties") or {} + required = [k for k in (js.get("required") or []) if k in props] + + unknown = [k for k in args if k not in props] + if not unknown: + try: + return schema(**args).model_dump(), None + except ValidationError as e: + raise ToolArgsError(describe_args(schema, args, e)) from e + + # 少了哪个必填的，就是多出来那个想写的；一个必填都不缺时，退而看还空着哪个可选字段 + missing = [k for k in required if k not in args] or [k for k in props if k not in args] + residual: ValidationError | None = None + if len(unknown) == 1 and len(missing) == 1: + fixed = dict(args) + fixed[missing[0]] = fixed.pop(unknown[0]) + try: + payload = schema(**fixed).model_dump() + except ValidationError as e: + # 改完名还是不合法。把这些毛病一并说出来，省得对方改对了名字 + # 再回来撞一次类型——describe_args 会跳过已经报过的那几个字段 + residual = e + else: + return payload, f"参数名 {unknown[0]} 不存在，已按唯一候选 {missing[0]} 执行" + + raise ToolArgsError(describe_args(schema, args, residual)) + + +async def call_tool( + name: str, + args: dict[str, Any], + ctx: ToolContext, + session: Any = None, + *, + on_fix: Callable[[str], None] | None = None, +) -> Any: + """直接调用一个工具（工具节点、以及设置页的"试一下"按钮用）。 + + 两条分支都先过 args_schema。动态工具那条以前是裸 `**args` 展开的，参数名写错 + 时抛的是 TypeError，报错里只剩内部闭包名——`_make_query_tool.._run() got + an unexpected keyword argument 'query'` 既指不到是哪个工具，也说不出该填什么。 + + on_fix 在参数名被纠正时回调一次。call_tool 不该知道事件系统长什么样， + 所以只把这件事交出去，由调用方决定说给谁听。 + """ spec = get_spec(name) if spec: - validated = spec.args_schema(**args) - payload = validated.model_dump() + payload, note = prepare_args(spec.args_schema, args) + if note and on_fix: + on_fix(note) if spec.needs_context: return await spec.func(ctx, **payload) return await spec.func(**payload) @@ -165,9 +310,18 @@ async def call_tool(name: str, args: dict[str, Any], ctx: ToolContext, session: if not tools: raise KeyError(f"找不到工具 {name!r}") tool = tools[0] + + model = args_model_of(tool) + if model is None: + payload = dict(args) + else: + payload, note = prepare_args(model, args) + if note and on_fix: + on_fix(note) + if tool.coroutine: - return await tool.coroutine(**args) - result = tool.invoke(args) + return await tool.coroutine(**payload) + result = tool.invoke(payload) if inspect.isawaitable(result): return await result return result diff --git a/backend/tests/test_agent_tool_loop.py b/backend/tests/test_agent_tool_loop.py new file mode 100644 index 00000000..39823e7a --- /dev/null +++ b/backend/tests/test_agent_tool_loop.py @@ -0,0 +1,180 @@ +"""Agent 的工具循环，跑真图验收。 + +单测能保证 split_tool_calls 分得对、prepare_args 纠得对，但保证不了它们在 +run_agent 里被接对了。这里从 run_manager.start 进去、从落库的 RunEvent 出来， +验的是用户真正会看到的那条事件流。 + +两件必须守住的事： + +- 每个 tool_use 都要有配对的 tool_result。少一条，下一次模型调用直接 400。 +- 没执行的调用绝不能发 tool.start。前端靠 call_id 配对 start/end + （frontend/src/run/decode.ts），一条没有结尾的 start 就是界面上一个永远转圈的卡片。 +""" + +from __future__ import annotations + +import asyncio +import sqlite3 + +import pytest +from langchain_core.messages import AIMessage, ToolMessage +from sqlalchemy import select + +from app.db.base import SessionLocal +from app.db.models import DataSource, Run, RunEvent +from app.engine.runner import run_manager + + +@pytest.fixture +async def runner(): + """真图要跑起来就得有 checkpointer —— 平时是 FastAPI 的 lifespan 建的。""" + await run_manager.setup() + yield run_manager + await run_manager.shutdown() + + +@pytest.fixture +async def shop(tmp_path): + """一个真实的小库 + 一条 DataSource 记录，工具要能真的查到数。""" + path = tmp_path / "shop.db" + db = sqlite3.connect(path) + db.executescript("CREATE TABLE orders (id INTEGER PRIMARY KEY, amount REAL NOT NULL);") + db.executemany("INSERT INTO orders(id,amount) VALUES(?,?)", [(i, i * 2.0) for i in range(1, 8)]) + db.commit() + db.close() + + name = f"shop_{tmp_path.name[-6:].replace('-', '_')}" + async with SessionLocal() as session: + session.add(DataSource( + name=name, kind="sqlite", database=str(path), readonly=True, + description="测试库", options={}, schema_cache={}, enabled=True, + )) + await session.commit() + return name, str(path) + + +def _graph(tool_name: str) -> dict: + return { + "nodes": [ + {"id": "in", "type": "input", "data": {"config": {}}}, + {"id": "bot", "type": "agent", "data": {"config": { + "prompt": "查一下订单总额", + "tools": [tool_name], + "approval": "never", # 审批会挂起，这里要一路跑到底 + "max_steps": 3, + }}}, + {"id": "out", "type": "output", "data": {"config": {}}}, + ], + "edges": [ + {"source": "in", "target": "bot"}, + {"source": "bot", "target": "out"}, + ], + } + + +async def _run_to_completion(graph: dict) -> list[RunEvent]: + run = await run_manager.start(graph=graph, input_payload={"question": "订单总额"}) + for _ in range(300): + 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("运行没有在 15 秒内结束") + + 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 + + +async def test_serial_loop_runs_one_tool_and_answers_every_call(runner, shop, monkeypatch) -> None: + """模型一轮发两个工具调用：只跑第一个，但两个都要有回音。 + + 顺带把第一个的参数名写错成 query —— 这正是那条 + "_run() got an unexpected keyword argument 'query'" 的来源。 + """ + name, _ = shop + tool_name = f"db_query__{name}" + + from app.providers import mock_model + + turns = {"n": 0} + + def _decide(self, messages): + turns["n"] += 1 + if turns["n"] == 1: + return AIMessage( + content="先查一下。", + tool_calls=[ + # 参数名故意写错：能被纠正成 sql 才算过 + {"name": tool_name, "args": {"query": "SELECT SUM(amount) AS total FROM orders"}, + "id": "call_a"}, + {"name": tool_name, "args": {"sql": "SELECT COUNT(*) FROM orders"}, + "id": "call_b"}, + ], + ) + return AIMessage(content="订单总额是 56。") + + monkeypatch.setattr(mock_model.MockChatModel, "_decide", _decide) + + events, run = await _run_to_completion(_graph(tool_name)) + kinds = [(e.type, e.data) for e in events] + + assert run.status == "succeeded", [k for k, _ in kinds] + + # 一轮只跑了一个工具 + starts = [d for t, d in kinds if t == "tool.start"] + assert len(starts) == 1, f"应该只有一次 tool.start，实际 {len(starts)} 次" + assert starts[0]["call_id"] == "call_a" + # 发出去的是纠正后的参数，不是模型原样那份 + assert "sql" in starts[0]["args"] and "query" not in starts[0]["args"] + + # 每个 tool.start 都有配对的结尾，界面上不会留下转圈的卡片 + ends = [d for t, d in kinds if t in ("tool.end", "tool.error")] + assert {d["call_id"] for d in ends} == {"call_a"} + + # 纠正这件事说出来了，没有安静地把事办了 + warns = [d["message"] for t, d in kinds if t == "log" and d.get("level") == "warn"] + assert any("query" in w and "sql" in w for w in warns), warns + + # 工具真的查到了数 + assert "56" in str([d.get("preview") for t, d in kinds if t == "tool.end"]) + + +async def test_bad_args_feed_the_schema_back_instead_of_dying(runner, shop, monkeypatch) -> None: + """参数全不认识时不执行，把"它接受什么"喂回去——模型下一步才改得对。""" + name, _ = shop + tool_name = f"db_query__{name}" + + from app.providers import mock_model + + seen: list[str] = [] + turns = {"n": 0} + + def _decide(self, messages): + turns["n"] += 1 + if turns["n"] == 1: + return AIMessage( + content="查一下。", + tool_calls=[{"name": tool_name, "args": {"foo": 1, "bar": 2}, "id": "call_x"}], + ) + # 第二轮：把上一轮喂回来的内容记下来，这就是模型能看到的全部依据 + seen.extend(m.content for m in messages if isinstance(m, ToolMessage)) + return AIMessage(content="好的。") + + monkeypatch.setattr(mock_model.MockChatModel, "_decide", _decide) + + events, run = await _run_to_completion(_graph(tool_name)) + assert run.status == "succeeded" + + assert seen, "参数错了也必须给模型一条 tool_result，否则下一轮直接 400" + text = seen[0] + assert "sql" in text and "必填" in text # 说清楚该填什么 + assert "unexpected keyword argument" not in text # 不再是那句天书 + + # 没执行的调用不发 tool.start，界面上不会留下转不完的卡片 + assert not [e for e in events if e.type == "tool.start"] + assert [e for e in events if e.type == "tool.error"] diff --git a/backend/tests/test_thinking.py b/backend/tests/test_thinking.py index 50b6cb86..076df1df 100644 --- a/backend/tests/test_thinking.py +++ b/backend/tests/test_thinking.py @@ -52,3 +52,103 @@ def test_plain_message_has_no_thinking() -> None: def test_empty_reasoning_is_not_reported() -> None: """空串不算思考，否则界面上会出现一行什么都没有的"思考"。""" assert thinking_text(AIMessageChunk(content="", additional_kwargs={"reasoning_content": ""})) == "" + + +# -------------------------------------------------------------------------- +# 交错思考：工具结果回来之后还能再想一段 +# -------------------------------------------------------------------------- +# +# 串行跑工具时这一条最要紧——每一步之间本来就该重新判断一次"看到这个结果 +# 之后下一步该干嘛"。没有它，模型只在一轮开头想一次，后面全是埋头执行。 + + +def _anthropic(model: str, **spec_kw): + import os + + from app.db.models import Provider + from app.providers.factory import ModelSpec, build_chat_model + + os.environ.setdefault("ANTHROPIC_API_KEY", "test-only") + provider = Provider(name="t", kind="anthropic", api_key=None, models=[], + extra=spec_kw.pop("extra", {}) or {}) + return build_chat_model(provider, ModelSpec(model=model, **spec_kw)) + + +def test_adaptive_models_need_no_beta_header() -> None: + """4.6 起 adaptive thinking 自带交错。 + + 这里不是"忘了加"——在 Opus 4.6 上手动模式压根没有交错思考，只有 adaptive + 有，所以多挂一个 header 不但没用，还会把调用改道 client.beta.messages。 + """ + for model in ("claude-opus-5", "claude-sonnet-5", "claude-opus-4-8", "claude-sonnet-4-6"): + chat = _anthropic(model) + assert chat.betas is None, model + assert chat.thinking == {"type": "adaptive", "display": "summarized"}, model + + +def test_the_previous_generation_gets_the_beta_header() -> None: + """Claude 4 / 4.5 要显式开。目录里没列它们，但模型名是放行的（可以手填）。""" + for model in ("claude-opus-4-5", "claude-sonnet-4-5", "claude-opus-4"): + assert _anthropic(model).betas == ["interleaved-thinking-2025-05-14"], model + # openrouter 那种带前缀的写法也要认得出来 + assert _anthropic("openrouter/anthropic/claude-sonnet-4-5").betas == [ + "interleaved-thinking-2025-05-14" + ] + + +def test_the_beta_header_comes_with_thinking_actually_enabled() -> None: + """光挂 header 是一条什么都不做的空设置。 + + 这一代默认根本没开 thinking——没有思考可交错，header 挂了也白挂。 + 第一版就是这么写的：betas 有了，thinking 是 None。 + """ + chat = _anthropic("claude-sonnet-4-5") + assert chat.thinking == {"type": "enabled", "budget_tokens": 4096} + assert chat.betas == ["interleaved-thinking-2025-05-14"] + + +def test_thinking_and_temperature_never_go_out_together() -> None: + """开了 thinking 再传 temperature 就是 400。 + + 4.6 起的模型压根不收采样参数，supports_sampling 已经挡住了；漏的正是 + 4/4.5 这一代——它们本来是收 temperature 的。 + """ + assert _anthropic("claude-sonnet-4-5", temperature=0.7).temperature is None + # 没开 thinking 的还照常收 + assert _anthropic("claude-haiku-4-5", temperature=0.7).temperature == 0.7 + assert _anthropic("claude-sonnet-4-5", thinking="off", temperature=0.7).temperature == 0.7 + + +def test_models_without_the_feature_are_left_alone() -> None: + """Haiku 4.5 不支持交错思考，给它挂 header 只是噪音。""" + assert _anthropic("claude-haiku-4-5").betas is None + + +def test_thinking_off_means_no_interleaving_either() -> None: + """都不让它想了，就别再开"想得更碎"的开关。""" + assert _anthropic("claude-sonnet-4-5", thinking="off").betas is None + + +def test_a_gateway_that_rejects_the_beta_can_opt_out() -> None: + """自建网关可能不认这个 beta，而它是在请求时才炸的——留一个关得掉的口子。 + + 关掉的只是交错，思考本身要留着：想要"别想了"的人写的是 thinking=off。 + """ + chat = _anthropic("claude-sonnet-4-5", extra={"interleaved_thinking": False}) + assert chat.betas is None + assert chat.thinking == {"type": "enabled", "budget_tokens": 4096} + + +def test_serial_tool_choice_stays_compatible_with_interleaving() -> None: + """交错思考要求 tool_choice 是 auto 或 none。 + + 串行模式给的是 {"type": "auto", "disable_parallel_tool_use": true} —— auto + 不是强制调用，两者并存没问题。这条盯的是以后有人把它改成 any/tool： + 那样会同时踩坏交错思考和 thinking 本身。 + """ + from app.providers.factory import bind_tools_safely + from app.tools.registry import ToolContext, all_specs, build_tool + + tools = [build_tool(next(iter(all_specs().values())), ToolContext())] + bound = bind_tools_safely(_anthropic("claude-sonnet-4-5"), tools, parallel=False) + assert bound.kwargs["tool_choice"]["type"] in ("auto", "none") diff --git a/backend/tests/test_tool_args.py b/backend/tests/test_tool_args.py new file mode 100644 index 00000000..5c6afb2d --- /dev/null +++ b/backend/tests/test_tool_args.py @@ -0,0 +1,375 @@ +"""工具参数的校验、纠错，以及 agent 一轮只跑一个工具。 + +起因是一条谁也看不懂的报错： + + 工具 db_query__shop 执行失败： + _make_query_tool.._run() got an unexpected keyword argument 'query' + +参数名写成了 query（它叫 sql），而动态工具那条调用路径当时是裸 `**args` 展开的， +于是 pydantic 的"缺少 sql 字段"漏成了一个只剩内部闭包名的 TypeError —— 既指不到 +是哪个工具，也说不出该填什么，人和模型都只能接着猜。 +""" + +from __future__ import annotations + +import sqlite3 +from types import SimpleNamespace + +import pytest +from langchain_core.messages import ToolMessage +from pydantic import BaseModel, Field + +from app.engine.nodes.llm import split_tool_calls +from app.tools.datasource import _make_query_tool, _make_schema_tool +from app.tools.registry import ( + ToolArgsError, + ToolContext, + args_model_of, + describe_args, + prepare_args, +) + + +class _QueryLike(BaseModel): + """和 datasource 的查询参数同形：一个必填 + 一个可选。""" + + sql: str = Field(description="要执行的 SQL，一次一条") + limit: int | None = Field(default=None, description="最多返回多少行，默认 1000") + + +class _AllOptional(BaseModel): + table: str | None = Field(default=None, description="表名；留空则列出所有表") + + +@pytest.fixture +def source(tmp_path): + path = tmp_path / "shop.db" + db = sqlite3.connect(path) + db.executescript("CREATE TABLE orders (id INTEGER PRIMARY KEY, amount REAL NOT NULL);") + db.executemany("INSERT INTO orders(id,amount) VALUES(?,?)", [(i, i * 1.5) for i in range(1, 6)]) + db.commit() + db.close() + return SimpleNamespace( + # id 必须和别的测试文件不同：engine 按 source.id 缓存连接池 + # （app/data/engine.py:121），撞了 id 就会连到人家那个库上—— + # 默认的文件顺序下碰巧不冲突，单跑一个文件或换个顺序就串了 + id="tool-args-src", name="shop", kind="sqlite", database=str(path), + host=None, port=None, username=None, password=None, options={}, + readonly=True, description="测试库", schema_cache={}, + ) + + +# -------------------------------------------------------------------------- +# prepare_args +# -------------------------------------------------------------------------- + + +def test_unique_candidate_is_corrected() -> None: + """多出来一个 query、少了一个必填的 sql —— 候选唯一，改名就是了。""" + payload, note = prepare_args(_QueryLike, {"query": "SELECT 1"}) + assert payload["sql"] == "SELECT 1" + assert note and "query" in note and "sql" in note + + +def test_correct_args_pass_through_without_a_note() -> None: + """本来就对的不该被说成"纠正过"，否则轨迹里全是没用的黄条。""" + payload, note = prepare_args(_QueryLike, {"sql": "SELECT 1", "limit": 10}) + assert (payload["sql"], payload["limit"]) == ("SELECT 1", 10) + assert note is None + + +def test_ambiguous_names_are_not_guessed() -> None: + """两个都不认识时改名就是碰运气，不如把参数表摊开。""" + with pytest.raises(ToolArgsError) as e: + prepare_args(_QueryLike, {"foo": "SELECT 1", "bar": 3}) + text = str(e.value) + assert "sql" in text and "limit" in text + assert "foo" in text and "bar" in text + + +def test_typo_on_all_optional_schema_is_not_silently_dropped() -> None: + """全可选的 schema 上，pydantic 默认会把认不出的键悄悄丢掉。 + + 真丢了的话 {"tabel": "orders"} 会"成功"执行一次什么都没查的调用 —— + 这种静默失败比报错难查得多。这里必须要么纠正、要么报错。 + """ + payload, note = prepare_args(_AllOptional, {"tabel": "orders"}) + assert payload["table"] == "orders" + assert note is not None + + +def test_wrong_type_names_the_field_and_the_expected_type() -> None: + """名字对但类型不对，不是"写错名字"，别顺手改名把错误盖掉。""" + with pytest.raises(ToolArgsError) as e: + prepare_args(_QueryLike, {"sql": "SELECT 1", "limit": "很多"}) + assert "limit 的值不合法，应为整数" in str(e.value) + + +def test_missing_required_says_what_is_required() -> None: + with pytest.raises(ToolArgsError) as e: + prepare_args(_QueryLike, {}) + assert "缺少必填参数 sql" in str(e.value) + + +def test_every_problem_is_reported_at_once() -> None: + """改对了名字再回来撞一次类型错误，等于让人分两趟改。""" + with pytest.raises(ToolArgsError) as e: + prepare_args(_QueryLike, {"query": "SELECT 1", "limit": "很多"}) + text = str(e.value) + assert "query 不是它的参数" in text + assert "limit 的值不合法" in text + + +def test_the_message_is_a_whole_sentence() -> None: + """调用方会把它直接拼在工具名后面（"工具 X：…"），不能缺主语。""" + for args in ({}, {"foo": 1, "bar": 2}, {"sql": "SELECT 1", "limit": "很多"}): + with pytest.raises(ToolArgsError) as e: + prepare_args(_QueryLike, args) + assert str(e.value).startswith("参数不对：") + + +# -------------------------------------------------------------------------- +# describe_args —— 这段话是模型下一步改对的唯一依据 +# -------------------------------------------------------------------------- + + +def test_description_carries_the_field_help_text() -> None: + text = describe_args(_QueryLike, {"query": "SELECT 1"}) + assert "要执行的 SQL" in text # Field(description=...) 要搬进来 + assert "必填" in text and "可选" in text + assert "字符串" in text and "整数" in text # 可选字段是 anyOf[int, null]，要摊开 + assert "query" in text # 说清楚"你传的是什么" + + +# -------------------------------------------------------------------------- +# 真实的数据源工具：复现开头那条报错 +# -------------------------------------------------------------------------- + + +async def test_the_original_query_typo_now_runs(source) -> None: + tool = _make_query_tool(source, ToolContext()) + schema = args_model_of(tool) + assert schema is not None + + payload, note = prepare_args(schema, {"query": "SELECT COUNT(*) AS c FROM orders"}) + assert note is not None + result = await tool.coroutine(**payload) + assert '"c"' in result or "c" in result + + +async def test_raw_expansion_is_what_used_to_explode(source) -> None: + """记录一下被修掉的是什么：绕开校验直接展开，抛的就是那句天书。""" + tool = _make_query_tool(source, ToolContext()) + with pytest.raises(TypeError) as e: + await tool.coroutine(query="SELECT 1") + assert "unexpected keyword argument" in str(e.value) + + +async def test_schema_tool_args_are_validated_too(source) -> None: + tool = _make_schema_tool(source) + schema = args_model_of(tool) + payload, note = prepare_args(schema, {"tabel": "orders"}) + assert payload["table"] == "orders" and note is not None + + +async def test_call_tool_end_to_end_on_the_path_that_broke(source) -> None: + """走 call_tool 这个真实入口（工具节点和设置页的"试一下"都走它）。 + + 以前这条路径对动态工具是裸 `**args` 展开的，报错止步于 TypeError； + 现在参数名能被纠正、纠正这件事能通过 on_fix 说出去。 + """ + from app.db.base import SessionLocal + from app.db.models import DataSource + from app.tools.registry import call_tool + + async with SessionLocal() as session: + session.add(DataSource( + name="shop", kind="sqlite", database=source.database, + readonly=True, description="测试库", options={}, schema_cache={}, enabled=True, + )) + await session.commit() + + notes: list[str] = [] + async with SessionLocal() as session: + out = await call_tool( + "db_query__shop", {"query": "SELECT COUNT(*) AS c FROM orders"}, + ToolContext(), session=session, on_fix=notes.append, + ) + assert "5" in str(out) + # 跑通了，但配置里那个错的参数名还在，必须留下痕迹而不是安静地把事办了 + assert len(notes) == 1 and "sql" in notes[0] + + async with SessionLocal() as session: + with pytest.raises(ToolArgsError) as e: + await call_tool("db_query__shop", {"foo": 1, "bar": 2}, ToolContext(), session=session) + assert "sql" in str(e.value) + assert "unexpected keyword argument" not in str(e.value) + + +def test_tools_without_a_pydantic_schema_are_let_through() -> None: + """MCP 这类工具的 args_schema 可能是一份 JSON Schema dict，校验不了就放行。""" + assert args_model_of(SimpleNamespace(args_schema={"type": "object"})) is None + assert args_model_of(SimpleNamespace(args_schema=None)) is None + + +# -------------------------------------------------------------------------- +# agent 一轮只跑一个工具 +# -------------------------------------------------------------------------- + + +def _calls() -> list[dict]: + return [ + {"name": "db_schema__shop", "args": {}, "id": "c1"}, + {"name": "db_query__shop", "args": {"sql": "SELECT 1"}, "id": "c2"}, + {"name": "db_query__shop", "args": {"sql": "SELECT 2"}, "id": "c3"}, + ] + + +def test_serial_runs_only_the_first() -> None: + run_calls, deferred = split_tool_calls(_calls(), parallel=False) + assert [c["id"] for c in run_calls] == ["c1"] + assert [c["id"] for c, _ in deferred] == ["c2", "c3"] + + +def test_every_deferred_call_still_gets_a_result() -> None: + """两家 API 都要求每个 tool_use 有一条配对的 tool_result，少一条下次调用就是 400。""" + calls = _calls() + run_calls, deferred = split_tool_calls(calls, parallel=False) + answered = {c["id"] for c in run_calls} | {m.tool_call_id for _, m in deferred} + assert answered == {c["id"] for c in calls} + assert all(isinstance(m, ToolMessage) and m.content for _, m in deferred) + + +def test_parallel_keeps_every_call() -> None: + run_calls, deferred = split_tool_calls(_calls(), parallel=True) + assert len(run_calls) == 3 and deferred == [] + + +def test_single_call_is_never_deferred() -> None: + run_calls, deferred = split_tool_calls(_calls()[:1], parallel=False) + assert len(run_calls) == 1 and deferred == [] + + +# -------------------------------------------------------------------------- +# 串行也要告诉模型，省得它白发一堆调用 +# -------------------------------------------------------------------------- + + +def _one_tool(): + from app.tools.registry import all_specs, build_tool + + return [build_tool(next(iter(all_specs().values())), ToolContext())] + + +def test_anthropic_gets_disable_parallel_tool_use() -> None: + """ChatAnthropic 把它翻成 tool_choice，不是原样的 parallel_tool_calls。 + + type 必须是 auto：强制某个工具（any/tool）和 thinking 不能并存， + 而 agent 节点默认是开着 thinking 的。 + """ + from langchain_anthropic import ChatAnthropic + + from app.providers.factory import bind_tools_safely + + model = ChatAnthropic(model="claude-opus-5", api_key="x", + thinking={"type": "adaptive", "display": "summarized"}) + choice = bind_tools_safely(model, _one_tool(), parallel=False).kwargs["tool_choice"] + assert choice == {"type": "auto", "disable_parallel_tool_use": True} + assert "tool_choice" not in bind_tools_safely(model, _one_tool(), parallel=True).kwargs + + +def test_openai_gets_parallel_tool_calls() -> None: + from langchain_openai import ChatOpenAI + + from app.providers.factory import bind_tools_safely + + model = ChatOpenAI(model="gpt-4o", api_key="x") + assert bind_tools_safely(model, _one_tool(), parallel=False).kwargs["parallel_tool_calls"] is False + assert "parallel_tool_calls" not in bind_tools_safely(model, _one_tool(), parallel=True).kwargs + + +def test_binding_survives_models_that_ignore_the_flag() -> None: + """mock 和各种兼容网关不认这个参数也不能炸——串行本来就有编排层兜底。""" + from app.providers.factory import bind_tools_safely + from app.providers.mock_model import MockChatModel + + model = MockChatModel(model_name="mock-fast") + assert bind_tools_safely(model, _one_tool(), parallel=False) is not None + # 没有工具时原样返回，不该凭空包一层 + assert bind_tools_safely(model, [], parallel=False) is model + + +# -------------------------------------------------------------------------- +# 步数用完时的收尾 +# -------------------------------------------------------------------------- + + +def test_the_skip_note_never_becomes_the_answer() -> None: + """真出过事：run dd9927e6 交给用户的"答案"是一句内部管道文案。 + + 本轮只执行了第一个工具。看到它的结果之后，再决定这一个还要不要调… + + 步数用完时，收尾取的是 messages[-1]——而串行模式下那很可能是一条给模型看的 + ToolMessage。用户看到的不是"没跑完"，是一句冒充结论的管道说明。 + """ + from langchain_core.messages import AIMessage, HumanMessage + + from app.engine.nodes.llm import SKIPPED_NOTE + + messages = [ + HumanMessage(content="Shop 有几个管理员？"), + AIMessage(content="我先看看表结构。"), + ToolMessage(content="表 user …", tool_call_id="c1"), + ToolMessage(content=SKIPPED_NOTE, tool_call_id="c2"), + ] + # 收尾该取的是模型自己说过的最后一句，不是消息数组的最后一条 + picked = next( + (t for m in reversed(messages) + if isinstance(m, AIMessage) and (t := m.content.strip())), + "", + ) + assert picked == "我先看看表结构。" + assert SKIPPED_NOTE not in picked + + +# -------------------------------------------------------------------------- +# 查表结构可以一次问好几张 +# -------------------------------------------------------------------------- + + +async def test_schema_tool_takes_several_tables_at_once(source) -> None: + """一次调用就是一步。逐张问 6 张表就是 6 步，一次 8 步的预算直接见底。""" + from app.data import introspect + + source.schema_cache = await introspect.introspect(source) + tool = _make_schema_tool(source) + + one = await tool.coroutine(table="orders") + assert "orders" in one + + many = await tool.coroutine(table=["orders", "orders"]) + assert many.count("表 orders") == 2 # 两张都描述了，不是只认第一张 + + +async def test_schema_tool_accepts_a_comma_separated_string(source) -> None: + """模型两种写法都会用。为此拒掉一次调用，白白浪费一步。""" + from app.data import introspect + + source.schema_cache = await introspect.introspect(source) + out = await _make_schema_tool(source).coroutine(table="orders, orders") + assert out.count("表 orders") == 2 + + +async def test_schema_tool_still_lists_everything_when_asked_for_nothing(source) -> None: + from app.data import introspect + + source.schema_cache = await introspect.introspect(source) + out = await _make_schema_tool(source).coroutine() + assert "张表" in out and "orders" in out + + +def test_copilot_is_told_not_to_pin_max_steps() -> None: + """Copilot 把 max_steps: 8 写进了生成的图，后端默认值根本轮不上。""" + from app.api.copilot import NODE_REFERENCE + + assert "不要写 max_steps" in NODE_REFERENCE diff --git a/frontend/src/canvas/nodeDefs.ts b/frontend/src/canvas/nodeDefs.ts index cf708567..ed963e56 100644 --- a/frontend/src/canvas/nodeDefs.ts +++ b/frontend/src/canvas/nodeDefs.ts @@ -143,10 +143,15 @@ export const NODE_DEFS: Record = { ], help: '需要确认时运行会暂停，等你在审批面板放行', }, + { + key: 'parallel_tools', label: '并行执行工具', type: 'switch', advanced: true, + help: '默认关闭：一轮只调一个工具，看到结果再想下一步。开启后一轮可发多个，更快，' + + '但这一批中间没有新的思考', + }, ...MODEL_FIELDS, ...COMMON_TAIL, ], - defaults: { max_steps: 8, approval: 'dangerous', tools: [] }, + defaults: { max_steps: 12, approval: 'dangerous', tools: [], parallel_tools: false }, }, supervisor: { type: 'supervisor', label: '多 Agent 协作', category: '模型', icon: Users,