feat(fc): FC loop 加 enforceLoop 计数 + per-tool trace + 可复现 bench

eval 数据背书(FC vs 文本:0.40→0.90 / 0.80→0.85,FC 稳且快近一倍)后补齐 FC 两块短板:
- _run_fc_loop 支持 enforceLoop 计数模式(terminate 前 enforce_tool 须调够 minCount,
  挡早退);per-tool 写 ctx.trace(前端步数/trace 面板能看到 FC 过程,之前缺)。
  改用 _FCReactTerm(Term 子类,apply 收 ctx)替代 Tool 以传 ctx。门控从 react.enforceLoop
  取计数(序列模式=orchestrator 仍不走 FC)。
- evals/bench_fc.py:可复现的 FC vs 文本对比(同 agent/同 golden/qwen-plus,各跑一遍 →
  default_judge 打分 → 对比表 + bench_fc_report.json)。改了 FC 一键重测。
回归 test_function_calling(15,+enforceLoop+trace)+ lambdagent 613 全绿。
解锁:dashscope 单体(批改/润色/试卷…用 enforceLoop 计数)现可安全设 FC 默认。

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

evals/bench_fc.py

@@ -0,0 +1,127 @@
+#!/usr/bin/env python3
+"""FC vs 文本路径 benchmark — 可复现地对比"原生 function-calling"对真实 agent 质量的影响。
+
+在内核里直接跑(绕过 HTTP 鉴权),对同一 golden 任务、同一 agent、同一模型(qwen-plus),
+一次开 nativeToolCalls 一次关,收产物 → default_judge 打分 → 出对比表。改了 FC 一键重测。
+
+用法:
+  python3 evals/bench_fc.py [golden/literature-mapper.yaml ...]   # 默认跑 literature-mapper
+  MODEL=qwen-plus MAXSTEPS=14 python3 evals/bench_fc.py
+要 ~/.agentpaas/.env 里的 DASHSCOPE_API_KEY + 本地 ollama(judge)。会真调 LLM、花钱。
+"""
+from __future__ import annotations
+
+import json
+import os
+import shutil
+import sys
+import tempfile
+import time
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "agentpaas", "src"))
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "lambdagent", "src"))
+
+# 加载 keys
+_envf = os.path.expanduser("~/.agentpaas/.env")
+if os.path.exists(_envf):
+    for line in open(_envf):
+        if "=" in line and not line.strip().startswith("#"):
+            k, v = line.strip().split("=", 1)
+            os.environ.setdefault(k.strip(), v.strip().strip('"').strip("'"))
+
+import sqlite3                                              # noqa: E402
+import yaml                                                # noqa: E402
+from lambdagent.fromconfig import from_config              # noqa: E402
+from lambdagent.core import Context                        # noqa: E402
+from lambdagent.builtin_tools.shell_tools import _set_cwd  # noqa: E402
+from lambdagent.builtin_tools._sandbox import set_sandbox_root  # noqa: E402
+from agentpaas.engine.agent_eval import EvalTask, run_eval_task  # noqa: E402
+from agentpaas.engine.pipeline import default_judge        # noqa: E402
+
+MODEL = os.environ.get("MODEL", "qwen-plus")
+MAXSTEPS = int(os.environ.get("MAXSTEPS", "14"))
+DB = os.path.expanduser("~/.agentpaas/data/agentpaas.db")
+
+
+def _agent_cfg_for(template: str) -> dict:
+    con = sqlite3.connect(DB)
+    con.row_factory = sqlite3.Row
+    a = con.execute("SELECT id,current_version FROM agents WHERE agent_template=? "
+                    "AND status='active' LIMIT 1", (template,)).fetchone()
+    if not a:
+        raise RuntimeError(f"没有 {template} 的已安装实例")
+    cfg = json.loads(con.execute("SELECT config FROM agent_versions WHERE agent_id=? AND version=?",
+                                 (a["id"], a["current_version"])).fetchone()["config"])
+    # 统一成 qwen-plus + 限步数,公平对比
+    cfg["model"] = {"provider": "dashscope", "name": MODEL, "temperature": 0.0, "maxTokens": 4096}
+    cfg.setdefault("react", {})
+    cfg["react"]["maxSteps"] = MAXSTEPS
+    cfg["react"]["toolTimeout"] = 60
+    return cfg
+
+
+def _run_path(cfg: dict, fc: bool, task: EvalTask) -> dict:
+    d = tempfile.mkdtemp(prefix="bench_")
+    _set_cwd(d); set_sandbox_root(d)
+    try:
+        c = json.loads(json.dumps(cfg))
+        c["react"]["nativeToolCalls"] = fc
+        with tempfile.NamedTemporaryFile("w", suffix=".yml", delete=False, encoding="utf-8") as f:
+            yaml.dump(c, f, allow_unicode=True); p = f.name
+        term = from_config(p); os.unlink(p)
+        ctx = Context(workspace_path=d, run_id="bench")
+        t0 = time.time(); out = term.apply(task.input, ctx); dt = time.time() - t0
+        files = {fn: open(os.path.join(d, fn), encoding="utf-8", errors="ignore").read()
+                 for fn in os.listdir(d) if os.path.isfile(os.path.join(d, fn))}
+        return {"output": str(out), "steps": len(getattr(ctx, "trace", []) or []),
+                "cost_usd": 0.0, "workspace_path": d, "_files": files, "_dt": dt,
+                "_term": term._name}
+    finally:
+        set_sandbox_root(None); shutil.rmtree(d, ignore_errors=True)
+
+
+def bench_task(cfg: dict, task: EvalTask) -> dict:
+    row = {}
+    for label, fc in [("text", False), ("fc", True)]:
+        holder = {}
+
+        def wrapped(t, _fc=fc, _h=holder):
+            r = _run_path(cfg, _fc, t); _h.update(r); return r
+        r = run_eval_task(task, run_fn=wrapped,
+                          collect_artifacts_fn=lambda ws, _h=holder: _h.get("_files", {}),
+                          judge_fn=default_judge)
+        row[label] = {"files_ok": r.files_ok, "judge": round(r.judge_score, 2),
+                      "passed": r.passed, "dt": round(holder.get("_dt", 0)), "term": holder.get("_term", "")}
+    return row
+
+
+def main():
+    paths = sys.argv[1:] or [os.path.join(os.path.dirname(__file__), "golden", "literature-mapper.yaml")]
+    print(f"FC vs 文本 benchmark | model={MODEL} maxSteps={MAXSTEPS}\n")
+    allrows = {}
+    for p in paths:
+        spec = yaml.safe_load(open(p, encoding="utf-8"))
+        cfg = _agent_cfg_for(spec["agent_template"])
+        for t in spec.get("tasks", []):
+            task = EvalTask(id=t["id"], input=t.get("input", ""),
+                            must_produce=t.get("must_produce", []),
+                            rubric=t.get("rubric", ""), threshold=float(t.get("threshold", 0.6)))
+            print(f"--- {spec['agent_template']} / {task.id} ---")
+            row = bench_task(cfg, task)
+            allrows[task.id] = row
+            for lbl in ("text", "fc"):
+                r = row[lbl]
+                print(f"  {lbl:5} files_ok={r['files_ok']} judge={r['judge']:.2f} "
+                      f"passed={r['passed']} {r['dt']}s")
+            print()
+    print("=== 汇总(judge: 文本 → FC) ===")
+    for tid, row in allrows.items():
+        print(f"  {tid:20} {row['text']['judge']:.2f} → {row['fc']['judge']:.2f}  "
+              f"{'FC↑' if row['fc']['judge'] > row['text']['judge'] else ''}")
+    out = os.path.join(os.path.dirname(__file__), "bench_fc_report.json")
+    json.dump(allrows, open(out, "w", encoding="utf-8"), ensure_ascii=False, indent=2)
+    print(f"\n报告:{out}")
+
+
+if __name__ == "__main__":
+    main()

evals/bench_fc_report.json

@@ -0,0 +1,18 @@
+{
+  "sla-lit-map": {
+    "text": {
+      "files_ok": true,
+      "judge": 0.8,
+      "passed": true,
+      "dt": 135,
+      "term": "Memory(Loop(文献地图助手.react_step))"
+    },
+    "fc": {
+      "files_ok": true,
+      "judge": 0.85,
+      "passed": true,
+      "dt": 77,
+      "term": "Memory(文献地图助手.fc_react)"
+    }
+  }
+}

lambdagent/src/lambdagent/fromconfig/compiler.py

@@ -777,10 +777,12 @@ def _tools_json_schema(cfg: Dict) -> list:
 
 def _run_fc_loop(provider, system_prompt: str, user_input: str, tools: Dict,
                  tools_schema: list, *, max_steps: int, tool_timeout: int,
-                 on_step=None, cancel=None) -> str:
-    """原生 function-calling 的 ReAct 循环(阶段 2)。用结构化 tool_calls 直接执行,
-    不解析文本——根治 0 执行/格式飘/路径别名类 bug。见 docs/NATIVE_FUNCTION_CALLING_DESIGN.md。
-    复用现有工具执行(_timeout_call)+ on_step 事件(StepEvent),与文本 react 同形。"""
+                 on_step=None, cancel=None, ctx=None,
+                 enforce_tool: str = "", enforce_min: int = 0) -> str:
+    """原生 function-calling 的 ReAct 循环。用结构化 tool_calls 直接执行,不解析文本——
+    根治 0 执行/格式飘/路径别名类 bug。见 docs/NATIVE_FUNCTION_CALLING_DESIGN.md。
+    复用 _timeout_call 工具执行 + on_step(StepEvent);per-tool 写 ctx.trace(可观测);
+    支持 enforceLoop 计数模式(terminate 前 enforce_tool 须调够 enforce_min 次)。"""
     try:
         from lambdagent.agentruntime.react_engine import (
             StepEvent, STEP_THINK, STEP_TOOL_CALL, STEP_TOOL_RESULT)
@@ -794,8 +796,10 @@ def _run_fc_loop(provider, system_prompt: str, user_input: str, tools: Dict,
             except Exception:
                 pass
 
+    import time as _time
     messages = [{"role": "system", "content": system_prompt},
                 {"role": "user", "content": user_input}]
+    tool_counts: Dict[str, int] = {}
     final = ""
     for step in range(max_steps):
         if cancel is not None and cancel.is_cancelled():
@@ -807,8 +811,7 @@ def _run_fc_loop(provider, system_prompt: str, user_input: str, tools: Dict,
             _emit(STEP_THINK, step, content)
         if not tcs:
             final = content or final
-            break  # 模型给了最终答案、无工具调用 → 结束
-        # 回放 assistant 的 tool_calls(OpenAI 协议要求)
+            break
         messages.append({"role": "assistant", "content": content or "",
             "tool_calls": [{"id": tc["id"] or f"call_{step}_{i}", "type": "function",
                             "function": {"name": tc["name"],
@@ -819,11 +822,20 @@ def _run_fc_loop(provider, system_prompt: str, user_input: str, tools: Dict,
             name, args = tc["name"], (tc["arguments"] if isinstance(tc["arguments"], dict) else {})
             tcid = tc["id"] or f"call_{step}_{i}"
             if name == "terminate":
+                # enforceLoop 计数模式:enforce_tool 没调够 enforce_min 次 → 禁止结束。
+                got = tool_counts.get(enforce_tool, 0)
+                if enforce_tool and enforce_min > 0 and got < enforce_min:
+                    msg = (f"[SYSTEM] 还不能结束:{enforce_tool} 已调用 {got}/{enforce_min} 次,"
+                           f"还需 {enforce_min - got} 次。请继续完成后再 terminate。")
+                    messages.append({"role": "tool", "tool_call_id": tcid, "content": msg})
+                    _emit(STEP_TOOL_RESULT, step, msg, tool="terminate")
+                    continue  # 不 terminate,循环继续
                 final = (args.get("summary") or "").strip() or content or final
                 terminated = True
                 messages.append({"role": "tool", "tool_call_id": tcid, "content": "ok"})
                 continue
             _emit(STEP_TOOL_CALL, step, json.dumps(args, ensure_ascii=False), tool=name)
+            _t0 = _time.time()
             tool = tools.get(name)
             if tool is None:
                 obs = f"[TOOL_ERROR] unknown tool: {name}"
@@ -832,7 +844,13 @@ def _run_fc_loop(provider, system_prompt: str, user_input: str, tools: Dict,
                     obs = str(_timeout_call(tool, args, tool_timeout))
                 except Exception as e:
                     obs = f"[TOOL_ERROR] {e}"
+            tool_counts[name] = tool_counts.get(name, 0) + 1
             _emit(STEP_TOOL_RESULT, step, obs, tool=name)
+            if ctx is not None:  # per-tool 写 trace → 前端步数/trace 面板可见 FC 过程
+                try:
+                    ctx.log(f"fc:{name}", "", args, obs[:500], (_time.time() - _t0) * 1000)
+                except Exception:
+                    pass
             messages.append({"role": "tool", "tool_call_id": tcid,
                              "content": obs[:_MAX_OBS_LENGTH]})
         if terminated:
@@ -840,6 +858,23 @@ def _run_fc_loop(provider, system_prompt: str, user_input: str, tools: Dict,
     return final
 
 
+class _FCReactTerm(Term):
+    """FC react 车道的 Term(apply 收 ctx → 能写 trace)。比 Tool 多带 ctx 与 enforceLoop。"""
+    def __init__(self, name, *, provider, system_prompt, tools, tools_schema,
+                 max_steps, tool_timeout, on_step, cancel, enforce_tool, enforce_min):
+        super().__init__(name)
+        self._p = provider; self._sys = system_prompt; self._tools = tools
+        self._schema = tools_schema; self._ms = max_steps; self._tt = tool_timeout
+        self._os = on_step; self._c = cancel
+        self._etool = enforce_tool; self._emin = enforce_min
+
+    def apply(self, input, ctx=None):
+        return _run_fc_loop(self._p, self._sys, str(input), self._tools, self._schema,
+                            max_steps=self._ms, tool_timeout=self._tt, on_step=self._os,
+                            cancel=self._c, ctx=ctx, enforce_tool=self._etool,
+                            enforce_min=self._emin)
+
+
 def _compile_react(cfg: Dict, overrides: Dict) -> Term:
     """
     type: react -> Loop(react_step, condition, max_steps)
@@ -894,17 +929,15 @@ def _compile_react(cfg: Dict, overrides: Dict) -> Term:
                 from lambdagent.agentruntime import cancel as _fc_cancel
             except Exception:
                 _fc_cancel = None
-            _fc_schema = _tools_json_schema(cfg)
-            _fc_sys = cfg.get("systemPrompt", "")
-            _fc_on_step = overrides.get("on_step")
-
-            def _fc_apply(user_input, _p=_prov, _sys=_fc_sys, _tools=tools,
-                          _schema=_fc_schema, _ms=max_steps, _tt=tool_timeout,
-                          _os=_fc_on_step, _c=_fc_cancel):
-                return _run_fc_loop(_p, _sys, str(user_input), _tools, _schema,
-                                    max_steps=_ms, tool_timeout=_tt, on_step=_os, cancel=_c)
-
-            fc_term = Tool(f"{agent_name}.fc_react", _fc_apply)
+            # enforceLoop 计数模式(单体 agent 用):terminate 前 tool 须调够 minCount 次。
+            # 序列模式(orchestrator)FC 暂不支持——那些 agent 不设 nativeToolCalls。
+            _el = react_cfg.get("enforceLoop", {}) or {}
+            fc_term = _FCReactTerm(
+                f"{agent_name}.fc_react", provider=_prov, system_prompt=cfg.get("systemPrompt", ""),
+                tools=tools, tools_schema=_tools_json_schema(cfg),
+                max_steps=max_steps, tool_timeout=tool_timeout,
+                on_step=overrides.get("on_step"), cancel=_fc_cancel,
+                enforce_tool=_el.get("tool", ""), enforce_min=int(_el.get("minCount", 0) or 0))
             fc_term._think_ref = think  # token/usage 会计沿用 _find_usage_source
             return fc_term
 

lambdagent/tests/test_function_calling.py

@@ -160,3 +160,33 @@ class TestFCRouting(unittest.TestCase):
         from lambdagent.fromconfig.compiler import _compile_react
         term = _compile_react(self._cfg(False), {})
         self.assertFalse(term._name.endswith("fc_react"))
+
+
+class TestFCEnforceLoop(unittest.TestCase):
+    def test_terminate_blocked_until_min_count(self):
+        from lambdagent.fromconfig.compiler import _run_fc_loop
+        calls = []
+        tools = {"WriteFile": lambda a: calls.append(a) or "ok"}
+        prov = _FakeFCProvider([
+            {"content": None, "tool_calls": [{"id": "t1", "name": "terminate", "arguments": {"summary": "早退"}}]},
+            {"content": None, "tool_calls": [{"id": "w1", "name": "WriteFile", "arguments": {"file_path": "/a", "content": "x"}}]},
+            {"content": None, "tool_calls": [{"id": "t2", "name": "terminate", "arguments": {"summary": "完成"}}]},
+        ])
+        out = _run_fc_loop(prov, "s", "q", tools, [], max_steps=6, tool_timeout=10,
+                           enforce_tool="WriteFile", enforce_min=1)
+        self.assertEqual(len(calls), 1)   # 第一次 terminate 被挡 → 先调了 WriteFile
+        self.assertEqual(out, "完成")      # 满足后第二次 terminate 才放行
+
+
+class TestFCTrace(unittest.TestCase):
+    def test_logs_per_tool_to_ctx_trace(self):
+        from lambdagent.fromconfig.compiler import _run_fc_loop
+        from lambdagent.core import Context
+        ctx = Context()
+        prov = _FakeFCProvider([
+            {"content": None, "tool_calls": [{"id": "w1", "name": "WriteFile", "arguments": {"file_path": "/a", "content": "x"}}]},
+            {"content": None, "tool_calls": [{"id": "t1", "name": "terminate", "arguments": {"summary": "d"}}]},
+        ])
+        _run_fc_loop(prov, "s", "q", {"WriteFile": lambda a: "ok"}, [],
+                     max_steps=4, tool_timeout=10, ctx=ctx)
+        assert any(e.term_name == "fc:WriteFile" for e in ctx.trace)  # per-tool trace 写了