Runtime、工具注册与公共评测平台

mkdir platform-demo
cd platform-demo

"""运行时把工具名映射到函数，再核对参数、扣额度并记录结果。"""


class Runtime:
    def __init__(self, budget):
        if type(budget) is not int or budget < 0:
            raise ValueError("额度需要非负整数")
        self.remaining = budget
        self.tools = {}
        self.trace = []

    def register(self, name, function):
        # 注册阶段固定工具名，避免后注册的函数覆盖已审核实现。
        if not isinstance(name, str) or not name or name in self.tools or not callable(function):
            raise ValueError("工具名需要唯一，实现需要可调用")
        self.tools[name] = function

    def call(self, name, arguments):
        if not isinstance(name, str) or name not in self.tools:
            raise ValueError("工具未注册")
        # 这个注册表中的工具统一接受一个非空 query 字符串。
        if (type(arguments) is not dict or set(arguments) != {"query"}
                or not isinstance(arguments["query"], str) or not arguments["query"].strip()):
            raise ValueError("参数需要一个非空 query 字符串")
        if self.remaining == 0:
            raise ValueError("调用额度已用完")
        self.remaining -= 1  # 每次尝试消耗额度，包括实际执行失败。
        entry = {"tool": name, "status": "running"}
        self.trace.append(entry)
        try:
            result = self.tools[name](arguments["query"].strip())
            entry["status"] = "ok"
            return result
        except Exception:
            entry["status"] = "error"
            raise


def search(query):
    # 实际在固定资料列表中查询，结果可由读者逐项核对。
    documents = ["退款说明", "物流说明"]
    return [document for document in documents if query in document]


def benchmark(budget=2):
    cases = [("退款", ["退款说明"]), ("物流", ["物流说明"])]
    outcomes = []
    for query, expected in cases:
        # 每道任务重新创建运行时，避免上一道消耗的额度污染下一道。
        runtime = Runtime(budget)
        runtime.register("search", search)
        try:
            outcomes.append(runtime.call("search", {"query": query}) == expected)
        except ValueError:
            outcomes.append(False)
    return outcomes


def main():
    runtime = Runtime(2)
    runtime.register("search", search)
    print("查询结果：", runtime.call("search", {"query": "退款"}))
    print("剩余额度：", runtime.remaining)
    print("执行轨迹：", runtime.trace)
    outcomes = benchmark()
    print(f"基准任务通过：{sum(outcomes)}/{len(outcomes)}")


if __name__ == "__main__":
    main()

ls
python -X utf8 platform_demo.py
python -X utf8 -m unittest -v test_platform_demo.py

查询结果： ['退款说明']
剩余额度： 1
执行轨迹： [{'tool': 'search', 'status': 'ok'}]
基准任务通过：2/2

python -X utf8 platform_demo.py