Tool Evaluation — Đánh giá hiệu quả tools trong agent systems

Xây dựng agent với tools không khó. Khó là biết agent của bạn đang hoạt động tốt đến đâu. Khi nào Claude chọn đúng tool? Khi nào nó hallucinate tool parameters? Khi nào nó gọi tool không cần thiết? Không có systematic evaluation, bạn chỉ đang đoán mò.

Bài này trình bày framework đầy đủ để đánh giá tool effectiveness — từ định nghĩa metrics đến automated test suites và continuous monitoring.

Tại sao Tool Evaluation quan trọng?

Agent failures thường không rõ ràng như code errors. Agent có thể:

• Gọi đúng tool nhưng với wrong parameters — kết quả sai nhưng không crash
• Không gọi tool khi nên gọi — trả lời từ knowledge cũ, không accurate
• Gọi tool không cần thiết — tốn tokens và latency
• Gọi tool sai — confused giữa search vs calculate vs lookup

Chỉ với systematic evaluation bạn mới phát hiện và fix những vấn đề này.

Framework: 5 Dimensions of Tool Evaluation

from dataclasses import dataclass, field
from typing import Optional, Callable
import anthropic
import json
import time

client = anthropic.Anthropic()

@dataclass
class ToolCallExpectation:
    """Kỳ vọng về một tool call cụ thể"""
    tool_name: str                          # Tool phải được gọi
    required_params: dict = field(default_factory=dict)    # Params bắt buộc và giá trị mong đợi
    forbidden_params: list = field(default_factory=list)   # Params không được có
    param_validators: dict = field(default_factory=dict)  # Custom validators (key: validator_fn)

@dataclass
class TestCase:
    """Một test case cho agent evaluation"""
    id: str
    user_message: str
    expected_tool_calls: list = field(default_factory=list)  # List[ToolCallExpectation]
    should_not_call_tools: bool = False      # Đôi khi agent KHÔNG nên gọi tool
    expected_output_contains: list = field(default_factory=list)  # Keywords trong final response
    max_tool_calls: int = 5                  # Không gọi quá nhiều tools
    max_latency_ms: int = 10000             # Latency budget

@dataclass
class EvalResult:
    """Kết quả evaluation của một test case"""
    test_id: str
    passed: bool
    score: float  # 0.0 - 1.0
    tool_precision: float   # Correct tools called / Total tools called
    tool_recall: float      # Correct tools called / Expected tools
    param_accuracy: float   # Correct params / Total params
    latency_ms: int
    tool_calls_made: list
    issues: list
    response: str

Core Evaluation Engine

class ToolEvaluator:
    def __init__(self, tools: list, system_prompt: str = ""):
        self.tools = tools
        self.system_prompt = system_prompt
        self.client = anthropic.Anthropic()

    def evaluate_test_case(self, test: TestCase) -> EvalResult:
        """Chạy một test case và return detailed results"""
        start_time = time.time()
        tool_calls_made = []
        messages = [{"role": "user", "content": test.user_message}]
        final_response = ""
        issues = []

        # Run agent loop
        iteration = 0
        while iteration < test.max_tool_calls + 1:
            response = self.client.messages.create(
                model="claude-haiku-4-5",
                max_tokens=2000,
                system=self.system_prompt,
                tools=self.tools,
                messages=messages
            )

            if response.stop_reason == "end_turn":
                final_response = next((b.text for b in response.content if b.type == "text"), "")
                break

            elif response.stop_reason == "tool_use":
                # Record tool calls
                tool_results = []
                for block in response.content:
                    if block.type == "tool_use":
                        tool_call = {
                            "name": block.name,
                            "input": block.input,
                            "id": block.id
                        }
                        tool_calls_made.append(tool_call)

                        # Mock tool execution
                        result = self._mock_tool_execution(block.name, block.input)
                        tool_results.append({
                            "type": "tool_result",
                            "tool_use_id": block.id,
                            "content": json.dumps(result)
                        })

                messages.append({"role": "assistant", "content": response.content})
                messages.append({"role": "user", "content": tool_results})
                iteration += 1

            else:
                break

        latency_ms = int((time.time() - start_time) * 1000)

        # Check latency
        if latency_ms > test.max_latency_ms:
            issues.append(f"Latency {latency_ms}ms exceeded budget {test.max_latency_ms}ms")

        # Evaluate results
        scores = self._score_results(
            test, tool_calls_made, final_response, latency_ms, issues
        )

        return EvalResult(
            test_id=test.id,
            passed=scores["passed"],
            score=scores["overall_score"],
            tool_precision=scores["precision"],
            tool_recall=scores["recall"],
            param_accuracy=scores["param_accuracy"],
            latency_ms=latency_ms,
            tool_calls_made=tool_calls_made,
            issues=issues,
            response=final_response
        )

    def _score_results(self, test: TestCase, tool_calls: list,
                       response: str, latency_ms: int, issues: list) -> dict:
        """Tính điểm cho test case"""

        # Check: should NOT call tools
        if test.should_not_call_tools and tool_calls:
            issues.append(f"Called tools when shouldn't: {[tc['name'] for tc in tool_calls]}")
            return {"passed": False, "overall_score": 0.0, "precision": 0.0, "recall": 0.0, "param_accuracy": 0.0}

        if not test.expected_tool_calls:
            # No expectations about tools — only check response content
            content_ok = all(kw.lower() in response.lower() for kw in test.expected_output_contains)
            return {
                "passed": content_ok,
                "overall_score": 1.0 if content_ok else 0.5,
                "precision": 1.0, "recall": 1.0, "param_accuracy": 1.0
            }

        # Tool precision: out of all tools called, how many were expected?
        called_names = [tc["name"] for tc in tool_calls]
        expected_names = [e.tool_name for e in test.expected_tool_calls]

        correct_calls = [name for name in called_names if name in expected_names]
        precision = len(correct_calls) / len(called_names) if called_calls else 1.0
        recall = len(correct_calls) / len(expected_names) if expected_names else 1.0

        # Unexpected tools
        unexpected = [name for name in called_names if name not in expected_names]
        if unexpected:
            issues.append(f"Unexpected tool calls: {unexpected}")

        # Missing tools
        missing = [name for name in expected_names if name not in called_calls]
        if missing:
            issues.append(f"Missing expected tool calls: {missing}")

        # Parameter accuracy
        param_scores = []
        for expected in test.expected_tool_calls:
            actual_call = next((tc for tc in tool_calls if tc["name"] == expected.tool_name), None)
            if not actual_call:
                param_scores.append(0.0)
                continue

            actual_input = actual_call.get("input", {})
            param_score = self._score_params(expected, actual_input, issues)
            param_scores.append(param_score)

        param_accuracy = sum(param_scores) / len(param_scores) if param_scores else 1.0

        # Output content check
        content_score = 1.0
        if test.expected_output_contains:
            matched = sum(1 for kw in test.expected_output_contains if kw.lower() in response.lower())
            content_score = matched / len(test.expected_output_contains)
            if content_score < 1.0:
                missing_kw = [kw for kw in test.expected_output_contains if kw.lower() not in response.lower()]
                issues.append(f"Response missing keywords: {missing_kw}")

        overall_score = (precision * 0.3 + recall * 0.3 + param_accuracy * 0.3 + content_score * 0.1)
        passed = overall_score >= 0.8 and not any("Missing expected" in i for i in issues)

        return {
            "passed": passed,
            "overall_score": round(overall_score, 3),
            "precision": round(precision, 3),
            "recall": round(recall, 3),
            "param_accuracy": round(param_accuracy, 3)
        }

    def _score_params(self, expected: ToolCallExpectation, actual: dict, issues: list) -> float:
        """Score parameter correctness"""
        scores = []

        for param, expected_val in expected.required_params.items():
            if param not in actual:
                issues.append(f"Missing param '{param}' in {expected.tool_name}")
                scores.append(0.0)
            elif expected_val is not None and actual[param] != expected_val:
                issues.append(f"Wrong '{param}': expected '{expected_val}', got '{actual[param]}'")
                scores.append(0.5)  # Partial credit — param exists but wrong value
            else:
                scores.append(1.0)

        # Check forbidden params
        for param in expected.forbidden_params:
            if param in actual:
                issues.append(f"Forbidden param '{param}' used in {expected.tool_name}")
                scores.append(0.0)

        # Custom validators
        for param, validator in expected.param_validators.items():
            if param in actual:
                if not validator(actual[param]):
                    issues.append(f"Param '{param}' failed validation in {expected.tool_name}")
                    scores.append(0.0)

        return sum(scores) / len(scores) if scores else 1.0

    def _mock_tool_execution(self, tool_name: str, tool_input: dict) -> dict:
        """Mock tool results cho evaluation purposes"""
        return {"status": "success", "result": f"Mock result for {tool_name}", "input_received": tool_input}

Test Suite Builder

class ToolTestSuite:
    def __init__(self, name: str, evaluator: ToolEvaluator):
        self.name = name
        self.evaluator = evaluator
        self.test_cases = []

    def add_test(self, test: TestCase):
        self.test_cases.append(test)

    def run(self) -> dict:
        """Chạy toàn bộ test suite"""
        results = []
        print(f"
Running: {self.name} ({len(self.test_cases)} tests)")
        print("=" * 60)

        for test in self.test_cases:
            result = self.evaluator.evaluate_test_case(test)
            results.append(result)
            status = "PASS" if result.passed else "FAIL"
            print(f"  [{status}] {test.id} — score: {result.score:.2f} | latency: {result.latency_ms}ms")
            if result.issues:
                for issue in result.issues[:2]:
                    print(f"       Issue: {issue}")

        return self._aggregate_results(results)

    def _aggregate_results(self, results: list) -> dict:
        passed = sum(1 for r in results if r.passed)
        total = len(results)

        report = {
            "suite": self.name,
            "total": total,
            "passed": passed,
            "failed": total - passed,
            "pass_rate": round(passed / total * 100, 1) if total > 0 else 0,
            "avg_score": round(sum(r.score for r in results) / total, 3) if total > 0 else 0,
            "avg_precision": round(sum(r.tool_precision for r in results) / total, 3) if total > 0 else 0,
            "avg_recall": round(sum(r.tool_recall for r in results) / total, 3) if total > 0 else 0,
            "avg_param_accuracy": round(sum(r.param_accuracy for r in results) / total, 3) if total > 0 else 0,
            "avg_latency_ms": round(sum(r.latency_ms for r in results) / total) if total > 0 else 0,
            "failures": [r for r in results if not r.passed]
        }

        print(f"
Results: {passed}/{total} passed ({report['pass_rate']}%)")
        print(f"  Avg precision: {report['avg_precision']:.3f}")
        print(f"  Avg recall: {report['avg_recall']:.3f}")
        print(f"  Avg param accuracy: {report['avg_param_accuracy']:.3f}")
        print(f"  Avg latency: {report['avg_latency_ms']}ms")
        return report

Ví dụ: Test Suite cho Weather Agent

weather_tools = [
    {
        "name": "get_weather",
        "description": "Get current weather for a city",
        "input_schema": {
            "type": "object",
            "properties": {
                "city": {"type": "string"},
                "units": {"type": "string", "enum": ["celsius", "fahrenheit"], "default": "celsius"}
            },
            "required": ["city"]
        }
    },
    {
        "name": "get_forecast",
        "description": "Get weather forecast for next N days",
        "input_schema": {
            "type": "object",
            "properties": {
                "city": {"type": "string"},
                "days": {"type": "integer", "minimum": 1, "maximum": 7}
            },
            "required": ["city", "days"]
        }
    }
]

evaluator = ToolEvaluator(tools=weather_tools, system_prompt="You are a weather assistant.")
suite = ToolTestSuite("Weather Agent Tests", evaluator)

# Test 1: Simple current weather
suite.add_test(TestCase(
    id="T001_current_weather",
    user_message="What's the weather in Hanoi right now?",
    expected_tool_calls=[
        ToolCallExpectation(
            tool_name="get_weather",
            required_params={"city": "Hanoi"},
            param_validators={"units": lambda v: v in ["celsius", "fahrenheit"]}
        )
    ],
    expected_output_contains=["Hanoi", "weather"]
))

# Test 2: Forecast request
suite.add_test(TestCase(
    id="T002_5day_forecast",
    user_message="Give me the 5-day forecast for Ho Chi Minh City",
    expected_tool_calls=[
        ToolCallExpectation(
            tool_name="get_forecast",
            required_params={"city": "Ho Chi Minh City", "days": 5}
        )
    ]
))

# Test 3: Should NOT call tools
suite.add_test(TestCase(
    id="T003_no_tool_needed",
    user_message="What's the difference between weather and climate?",
    should_not_call_tools=True,
    expected_output_contains=["weather", "climate"]
))

# Run it
report = suite.run()
print(f"
Final pass rate: {report['pass_rate']}%")

Continuous Evaluation trong CI/CD

import subprocess
import sys

def run_eval_in_ci(suite: ToolTestSuite, min_pass_rate: float = 0.9) -> bool:
    """Gate CI/CD pipeline dựa trên eval results"""
    report = suite.run()
    pass_rate = report["pass_rate"] / 100

    if pass_rate < min_pass_rate:
        print(f"
CI GATE FAILED: {pass_rate*100:.1f}% < {min_pass_rate*100:.1f}% required")
        print("Failing tests:")
        for failure in report["failures"]:
            print(f"  - {failure.test_id}: {failure.issues}")
        return False

    print(f"
CI GATE PASSED: {pass_rate*100:.1f}% >= {min_pass_rate*100:.1f}%")
    return True

# In CI pipeline
success = run_eval_in_ci(suite, min_pass_rate=0.85)
if not success:
    sys.exit(1)  # Fail the CI build

Phân tích Failure Patterns

def analyze_failures(results: list[EvalResult]) -> dict:
    """Phân tích patterns trong test failures để tìm root causes"""
    failures = [r for r in results if not r.passed]
    if not failures:
        return {"message": "No failures to analyze"}

    # Group by failure type
    param_errors = []
    wrong_tool = []
    missing_tool = []
    over_calling = []

    for f in failures:
        for issue in f.issues:
            if "Wrong" in issue or "Missing param" in issue:
                param_errors.append(f.test_id)
            elif "Unexpected tool" in issue:
                wrong_tool.append(f.test_id)
            elif "Missing expected" in issue:
                missing_tool.append(f.test_id)

    print("
Failure Analysis:")
    print(f"  Parameter errors: {len(param_errors)} cases ({param_errors[:3]})")
    print(f"  Wrong tool called: {len(wrong_tool)} cases")
    print(f"  Tool not called: {len(missing_tool)} cases")

    recommendations = []
    if param_errors:
        recommendations.append("Improve tool input_schema descriptions — add examples for each param")
    if wrong_tool:
        recommendations.append("Clarify tool descriptions — disambiguate similar tools more clearly")
    if missing_tool:
        recommendations.append("Add more trigger phrases in tool descriptions for missed tools")

    return {"recommendations": recommendations}

analyze_failures([])

Tổng kết

Tool Evaluation không phải nice-to-have — đây là requirement cho production agent systems. Framework này cung cấp:

• 5 metrics: precision, recall, param accuracy, latency, content quality
• Automated test suites chạy được trong CI/CD
• Failure analysis để tìm root causes và improve tool descriptions
• Pass/fail gates để prevent regressions khi update tools

Đọc thêm: Extended Thinking + Tool Use — cải thiện tool selection accuracy với reasoning trước khi action.

Bài viết liên quan

• Extended Thinking + Tool Use — Suy luận sâu kết hợp công cụ
• Xây dựng LLM Agent từ đầu — Reference Implementation
• Context Compaction — Tự động nén context cho conversations dài
• Claude cho Engineering: Chiến lược testing toàn diện
• Claude cho Data: Phân tích dữ liệu tự động