Tool Calling Agent：ReAct、工具错误和可验证执行

#!/usr/bin/env python3
"""Dependency-free protocol test for verifiable tool-calling agents.

This is a deterministic state-machine experiment, not an LLM benchmark and not
a reproduction of ReAct/BFCL/ToolSandbox scores.  It makes one subtle failure
observable: a tool may commit a side effect and then time out.  Retrying with a
new idempotency key can duplicate the side effect even when the final answer is
correct; reusing the same key plus checking final state avoids that false pass.
"""

from __future__ import annotations

import argparse
import hashlib
import json
from dataclasses import asdict, dataclass, field
from typing import Any


SEED = 60
EXPECTED_SAMPLE_ID = "sample-007"
EXPECTED_VALUE = 7.4


@dataclass(frozen=True)
class ToolResult:
    ok: bool
    code: str
    observation: dict[str, Any]
    retryable: bool = False
    committed: bool = False


@dataclass
class LabState:
    samples: dict[str, str] = field(
        default_factory=lambda: {"S-7": EXPECTED_SAMPLE_ID}
    )
    assay_runs: list[dict[str, Any]] = field(default_factory=list)
    idempotency_cache: dict[str, dict[str, Any]] = field(default_factory=dict)


class LabEnvironment:
    """Small typed tool environment with injectable failure semantics."""

    def __init__(self, failure_mode: str = "none") -> None:
        self.state = LabState()
        self.failure_mode = failure_mode
        self.failure_consumed = False
        self.trace: list[dict[str, Any]] = []

    def _record(self, call: dict[str, Any], result: ToolResult) -> ToolResult:
        self.trace.append(
            {
                "step": len(self.trace) + 1,
                "call": call,
                "result": asdict(result),
                "assay_run_count": len(self.state.assay_runs),
            }
        )
        return result

    def execute(self, call: dict[str, Any]) -> ToolResult:
        if not isinstance(call, dict) or set(call) != {"name", "arguments"}:
            return self._record(
                call,
                ToolResult(False, "INVALID_CALL", {"error": "expected name+arguments"}),
            )
        name = call["name"]
        arguments = call["arguments"]
        if name not in {"find_sample", "run_assay", "read_assay"}:
            return self._record(
                call,
                ToolResult(False, "UNKNOWN_TOOL", {"error": f"unknown tool: {name}"}),
            )
        if not isinstance(arguments, dict):
            return self._record(
                call,
                ToolResult(False, "INVALID_ARGUMENT", {"error": "arguments must be an object"}),
            )
        if name == "find_sample":
            return self._find_sample(call, arguments)
        if name == "run_assay":
            return self._run_assay(call, arguments)
        return self._read_assay(call, arguments)

    def _find_sample(self, call: dict[str, Any], args: dict[str, Any]) -> ToolResult:
        if set(args) != {"label"} or not isinstance(args.get("label"), str):
            return self._record(
                call,
                ToolResult(False, "INVALID_ARGUMENT", {"error": "label must be a string"}),
            )
        sample_id = self.state.samples.get(args["label"])
        if sample_id is None:
            return self._record(
                call,
                ToolResult(False, "NOT_FOUND", {"error": "sample not found"}),
            )
        return self._record(call, ToolResult(True, "OK", {"sample_id": sample_id}))

    def _run_assay(self, call: dict[str, Any], args: dict[str, Any]) -> ToolResult:
        required = {"sample_id", "idempotency_key"}
        if set(args) != required or not all(isinstance(args.get(k), str) for k in required):
            return self._record(
                call,
                ToolResult(
                    False,
                    "INVALID_ARGUMENT",
                    {"error": "sample_id and idempotency_key must be strings"},
                ),
            )
        if args["sample_id"] != EXPECTED_SAMPLE_ID:
            return self._record(
                call,
                ToolResult(False, "NOT_FOUND", {"error": "sample_id not found"}),
            )
        key = args["idempotency_key"]
        if key in self.state.idempotency_cache:
            cached = self.state.idempotency_cache[key]
            return self._record(
                call,
                ToolResult(True, "IDEMPOTENT_REPLAY", cached, committed=True),
            )
        if self.failure_mode == "timeout_before_commit" and not self.failure_consumed:
            self.failure_consumed = True
            return self._record(
                call,
                ToolResult(
                    False,
                    "TIMEOUT_BEFORE_COMMIT",
                    {"error": "no state change"},
                    retryable=True,
                ),
            )
        run = {
            "run_id": f"run-{len(self.state.assay_runs) + 1:03d}",
            "sample_id": args["sample_id"],
            "value": EXPECTED_VALUE,
            "idempotency_key": key,
        }
        self.state.assay_runs.append(run)
        observation = {"run_id": run["run_id"], "value": run["value"]}
        self.state.idempotency_cache[key] = observation
        if self.failure_mode == "timeout_after_commit" and not self.failure_consumed:
            self.failure_consumed = True
            return self._record(
                call,
                ToolResult(
                    False,
                    "TIMEOUT_UNKNOWN_COMMIT",
                    {"error": "caller cannot know whether commit happened"},
                    retryable=True,
                    committed=True,
                ),
            )
        return self._record(call, ToolResult(True, "OK", observation, committed=True))

    def _read_assay(self, call: dict[str, Any], args: dict[str, Any]) -> ToolResult:
        if set(args) != {"sample_id"} or not isinstance(args.get("sample_id"), str):
            return self._record(
                call,
                ToolResult(False, "INVALID_ARGUMENT", {"error": "sample_id must be a string"}),
            )
        runs = [r for r in self.state.assay_runs if r["sample_id"] == args["sample_id"]]
        if not runs:
            return self._record(
                call,
                ToolResult(False, "NOT_FOUND", {"error": "no assay result"}),
            )
        return self._record(
            call,
            ToolResult(True, "OK", {"value": runs[-1]["value"], "runs": len(runs)}),
        )


def trace_digest(trace: list[dict[str, Any]]) -> str:
    payload = json.dumps(trace, sort_keys=True, separators=(",", ":"))
    return hashlib.sha256(payload.encode("utf-8")).hexdigest()[:12]


def verify(env: LabEnvironment, answer: dict[str, Any] | None) -> dict[str, Any]:
    answer_correct = answer == {"sample_id": EXPECTED_SAMPLE_ID, "value": EXPECTED_VALUE}
    exactly_one_side_effect = len(env.state.assay_runs) == 1
    state_correct = exactly_one_side_effect and all(
        run["sample_id"] == EXPECTED_SAMPLE_ID for run in env.state.assay_runs
    )
    return {
        "answer_correct": answer_correct,
        "state_correct": state_correct,
        "side_effect_count": len(env.state.assay_runs),
        "task_pass": answer_correct and state_correct,
    }


def run_timeout_episode(reuse_idempotency_key: bool) -> dict[str, Any]:
    env = LabEnvironment("timeout_after_commit")
    found = env.execute({"name": "find_sample", "arguments": {"label": "S-7"}})
    sample_id = found.observation["sample_id"]
    base_key = "task-060-assay"
    first = env.execute(
        {
            "name": "run_assay",
            "arguments": {"sample_id": sample_id, "idempotency_key": base_key},
        }
    )
    assert first.code == "TIMEOUT_UNKNOWN_COMMIT" and first.retryable
    retry_key = base_key if reuse_idempotency_key else base_key + "-retry"
    env.execute(
        {
            "name": "run_assay",
            "arguments": {"sample_id": sample_id, "idempotency_key": retry_key},
        }
    )
    read = env.execute({"name": "read_assay", "arguments": {"sample_id": sample_id}})
    answer = {"sample_id": sample_id, "value": read.observation["value"]}
    result = verify(env, answer)
    result.update(
        {
            "episode": "verified_retry" if reuse_idempotency_key else "blind_retry",
            "tool_calls": len(env.trace),
            "trace_digest": trace_digest(env.trace),
        }
    )
    return result


def run_schema_repair_episode() -> dict[str, Any]:
    env = LabEnvironment()
    bad = env.execute({"name": "find_sample", "arguments": {"label": 7}})
    assert bad.code == "INVALID_ARGUMENT" and not bad.retryable
    repaired = env.execute({"name": "find_sample", "arguments": {"label": "S-7"}})
    result = {
        "episode": "schema_repair",
        "invalid_calls": 1,
        "recovered": repaired.ok,
        "side_effect_count": len(env.state.assay_runs),
        "tool_calls": len(env.trace),
        "task_pass": repaired.ok and len(env.state.assay_runs) == 0,
        "trace_digest": trace_digest(env.trace),
    }
    return result


def run_loop_guard_episode(step_limit: int = 4) -> dict[str, Any]:
    env = LabEnvironment()
    for _ in range(step_limit):
        result = env.execute({"name": "invent_tool", "arguments": {}})
        assert result.code == "UNKNOWN_TOOL"
    return {
        "episode": "loop_guard",
        "force_terminated": True,
        "tool_calls": len(env.trace),
        "side_effect_count": len(env.state.assay_runs),
        "task_pass": len(env.trace) == step_limit and len(env.state.assay_runs) == 0,
        "trace_digest": trace_digest(env.trace),
    }


def run(check_only: bool) -> None:
    print(f"seed={SEED} deterministic_state_machine=true")
    episodes = [
        run_timeout_episode(reuse_idempotency_key=True),
        run_timeout_episode(reuse_idempotency_key=False),
        run_schema_repair_episode(),
        run_loop_guard_episode(),
    ]
    for episode in episodes:
        print(json.dumps(episode, ensure_ascii=False, sort_keys=True))

    verified, blind, repaired, guarded = episodes
    assert verified["task_pass"] and verified["side_effect_count"] == 1
    assert blind["answer_correct"] and not blind["state_correct"]
    assert blind["side_effect_count"] == 2 and not blind["task_pass"]
    assert repaired["task_pass"] and repaired["recovered"]
    assert guarded["task_pass"] and guarded["force_terminated"]
    if check_only:
        print("check-only: PASS")


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser()
    parser.add_argument("--check-only", action="store_true")
    return parser.parse_args()


if __name__ == "__main__":
    args = parse_args()
    run(args.check_only)