Inspect AI Python 教程：Agent 工具调用轨迹、评分与执行限额

python3 -m venv .venv
.venv/bin/pip install 'inspect-ai==0.3.281'

"""Tested with inspect-ai==0.3.281. Uses only local synthetic order data."""
import json
from inspect_ai import Task, task
from inspect_ai.dataset import Sample
from inspect_ai.model import ChatMessageAssistant, ChatMessageTool
from inspect_ai.scorer import CORRECT, INCORRECT, Score, accuracy, scorer
from inspect_ai.solver import TaskState, generate, use_tools
from inspect_ai.tool import ToolError, tool

@tool
def lookup_order():
    async def execute(order_id: str) -> str:
        """Read the status of a synthetic order.

        Args:
            order_id: Synthetic order ID to query.
        """
        if order_id == "1003":
            raise ToolError("Synthetic service unavailable")
        orders = {"1001": "delivered", "1002": "cancelled"}
        if order_id not in orders:
            raise ToolError("Synthetic order not found")
        return json.dumps({"order_id": order_id, "status": orders[order_id]})
    return execute

def judge_trace(messages, completion, order_id, status):
    calls = [c for m in messages if isinstance(m, ChatMessageAssistant) for c in (m.tool_calls or [])]
    results = [m for m in messages if isinstance(m, ChatMessageTool)]
    expected = {"order_id": order_id, "status": status}
    flags = {"exact_call": False, "successful_result": False, "exact_answer": False}
    if len(calls) == 1:
        call = calls[0]
        flags["exact_call"] = (call.function == "lookup_order" and call.arguments == {"order_id": order_id} and not call.parse_error)
        if len(results) == 1:
            result = results[0]
            try:
                flags["successful_result"] = (result.tool_call_id == call.id and result.function == "lookup_order" and result.error is None and json.loads(result.text) == expected)
            except (TypeError, ValueError):
                pass
    try:
        flags["exact_answer"] = json.loads(completion) == expected
    except (TypeError, ValueError):
        pass
    return flags

@scorer(metrics=[accuracy()])
def trajectory_check():
    async def score(state: TaskState, target):
        flags = judge_trace(state.messages, state.output.completion, state.metadata["order_id"], target.text)
        return Score(value=CORRECT if all(flags.values()) else INCORRECT,
                     answer=state.output.completion, metadata=flags,
                     explanation=json.dumps(flags))
    return score

@task
def order_eval():
    return Task(
        dataset=[Sample(id="order-1001", input='Query synthetic order 1001 with lookup_order. Reply with only JSON containing order_id and status.', target="delivered", metadata={"order_id": "1001"})],
        solver=[use_tools(lookup_order()), generate()],
        scorer=trajectory_check(),
        message_limit=12, turn_limit=3, token_limit=2000, time_limit=30,
    )

import json
from pathlib import Path
from inspect_ai import eval
from inspect_ai.model import ChatCompletionChoice, ChatMessageAssistant, ModelOutput, get_model
from inspect_ai.tool import ToolCall
from tool_eval import order_eval

def call_output(calls):
    return ModelOutput(model="mockllm", choices=[ChatCompletionChoice(
        message=ChatMessageAssistant(content="", tool_calls=calls), stop_reason="tool_calls")])

def run_case(name, outputs, log_dir):
    model = get_model("mockllm/model", memoize=False, custom_outputs=outputs)
    log = eval(order_eval(), model=model, display="none", log_dir=str(log_dir),
               max_samples=1, retry_on_error=0)[0]
    if log.status != "success" or len(log.samples or []) != 1 or log.samples[0].error:
        raise AssertionError(f"{name}: evaluation execution failed")
    sample = log.samples[0]
    score = sample.scores["trajectory_check"]
    return {"case": name, "score": score.value, "flags": score.metadata,
            "log_file": log.location}

if __name__ == "__main__":
    import argparse
    parser = argparse.ArgumentParser()
    parser.add_argument("--log-dir", default="./inspect-fixture-logs")
    args = parser.parse_args()
    report = run_case("correct", [
        call_output([ToolCall(id="lookup-1", function="lookup_order", arguments={"order_id":"1001"})]),
        ModelOutput.from_content("mockllm", '{"order_id":"1001","status":"delivered"}')
    ], Path(args.log_dir))
    assert report["score"] == "C", report
    print(json.dumps(report, ensure_ascii=False, indent=2))

.venv/bin/python run_mock.py --log-dir ./inspect-fixture-logs
.venv/bin/inspect view --log-dir ./inspect-fixture-logs

.venv/bin/python -m unittest discover -s aimazing/assets/inspect-ai -p 'test_tool_eval.py' -v