Agent学习笔记： 如何验证模型的tool-using能力

import os
# os.environ["OPENAI_API_KEY"] = "..."
from openai import OpenAI
import json

client = OpenAI()

tools = [
    {
        "type": "function",
        "function": {
            "name": "get_medical_definition",
            "description": "Get the definition of a medical term",
            "parameters": {
                "type": "object",
                "properties": {
                    "term": {"type": "string", "description": "The medical term to define"}
                },
                "required": ["term"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "get_treatment_options",
            "description": "Get treatment options for a disease",
            "parameters": {
                "type": "object",
                "properties": {
                    "disease": {"type": "string", "description": "The disease to get treatment options for"}
                },
                "required": ["disease"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "get_medication_side_effects",
            "description": "Get the potential side effects of a medication",
            "parameters": {
                "type": "object",
                "properties": {
                    "medication": {"type": "string", "description": "The name of the medication"}
                },
                "required": ["medication"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "get_disease_symptoms",
            "description": "Get the symptoms associated with a specific disease",
            "parameters": {
                "type": "object",
                "properties": {
                    "disease": {"type": "string", "description": "The disease to get symptoms for"}
                },
                "required": ["disease"]
            }
        }
    }
]

test_cases = [
    {
        "query": "What are the symptoms of influenza?",
        "expected_function": "get_disease_symptoms",
        "function_arguments": {"disease": "influenza"},
        "expected_response": "Symptoms of influenza include fever, chills, muscle aches, cough, congestion, runny nose, headaches, and fatigue."
    },
    {
        "query": "What is the definition of asthma?",
        "expected_function": "get_medical_definition",
        "function_arguments": {"term": "asthma"},
        "expected_response": "Asthma is a condition in which your airways narrow and swell and may produce extra mucus, making breathing difficult and triggering coughing, wheezing, and shortness of breath."
    },
    {
        "query": "What treatments are available for rheumatoid arthritis?",
        "expected_function": "get_treatment_options",
        "function_arguments": {"disease": "rheumatoid arthritis"},
        "expected_response": "Treatment options for rheumatoid arthritis include medications such as NSAIDs, steroids, DMARDs, and biologics, as well as physical therapy and lifestyle changes."
    },
    {
        "query": "What are the side effects of metformin?",
        "expected_function": "get_medication_side_effects",
        "function_arguments": {"medication": "metformin"},
        "expected_response": "Common side effects of metformin include gastrointestinal issues such as diarrhea, nausea, and abdominal discomfort, as well as lactic acidosis in rare cases."
    },
    {
        "query": "What symptoms are associated with hypothyroidism?",
        "expected_function": "get_disease_symptoms",
        "function_arguments": {"disease": "hypothyroidism"},
        "expected_response": "Symptoms of hypothyroidism include fatigue, weight gain, cold intolerance, dry skin, constipation, and depression."
    },
    ...
  ]

def call_openai_tool(model, prompt, tools):
    response = client.chat.completions.create(
        model=model,
        messages=[
            {"role": "system", "content": "You are a helpful assistant."},
            {"role": "user", "content": prompt}
        ],
        tools=tools,
        tool_choice="auto"
    )
    return response


def run_tests(test_cases, model=deployment_name):
    passed_tests = 0
    failed_tests = 0

    for case in test_cases:
        response = call_openai_tool(model, case['query'], tools)

        if case['expected_function'] and case['function_arguments']:
            if response.choices[0].message.tool_calls:
                tool_calls = response.choices[0].message.tool_calls

                # 验证函数调用名称和参数
                try:
                    assert tool_calls[0].function.name == case['expected_function'], f"Expected function {case['expected_function']}, but got {tool_calls[0].function.name}"
                    assert json.loads(tool_calls[0].function.arguments) == case['function_arguments'], f"Expected arguments {case['function_arguments']}, but got {tool_calls[0].function.arguments}"

                    print(f"Test passed for query: {case['query']}")
                    passed_tests += 1
                except AssertionError as e:
                    print(e)
                    print(f"Test failed for query: {case['query']}")
                    failed_tests += 1
            else:
                print(f"Test failed for query (no tool call): {case['query']}")
                failed_tests += 1
        else:
            # # 验证响应内容
            # response_text = response.choices[0].message.content.strip()

            # 验证没有调用工具
            if response.choices[0].message.tool_calls:
                print(f"Test failed for query (unexpected tool call): {case['query']}")
                failed_tests += 1

            else:
                print(f"Test passed for query: {case['query']}")
                passed_tests += 1

    total_tests = passed_tests + failed_tests
    accuracy = (passed_tests / total_tests) * 100 if total_tests > 0 else 0
    print(f"Total tests: {total_tests}, Passed: {passed_tests}, Failed: {failed_tests}, Accuracy: {accuracy:.2f}%")


run_tests(test_cases)

Test passed for query: What are the symptoms of chronic kidney disease?
Test passed for query: What does the medical term 'bipolar disorder' mean?
Expected arguments {'disease': 'chronic obstructive pulmonary disease'}, but got {
  "disease": "COPD"
}
Test failed for query: What are the treatment options for chronic obstructive pulmonary disease (COPD)?
Test passed for query: What side effects are associated with the use of aspirin?
Test passed for query: Tell me a joke.
Test passed for query: What is the capital of France?
Total tests: 22, Passed: 21, Failed: 1, Accuracy: 95.45%

! pip install rouge-score --quiet 

from rouge_score import rouge_scorer

def calculate_rouge_l(predicted, expected):
    scorer = rouge_scorer.RougeScorer(['rougeL'], use_stemmer=True)
    scores = scorer.score(predicted, expected)
    return scores['rougeL'].fmeasure

def run_tests_with_rouge(test_cases, model=deployment_name, rouge_l_threshold=0.8):
    passed_tests = 0
    failed_tests = 0

    for case in test_cases:
        response = call_openai_tool(model, case['query'], tools)

        if case['expected_function'] and case['function_arguments']:
            if response.choices[0].message.tool_calls:
                tool_calls = response.choices[0].message.tool_calls

                # 验证函数调用名称和参数
                try:
                    assert tool_calls[0].function.name == case['expected_function'], f"Expected function {case['expected_function']}, but got {tool_calls[0].function.name}"
                    predicted_arguments = json.loads(tool_calls[0].function.arguments)
                    
                    # 计算ROUGE-L分数
                    rouge_l_score = calculate_rouge_l(json.dumps(predicted_arguments), json.dumps(case['function_arguments']))
                    
                    assert rouge_l_score >= rouge_l_threshold, f"ROUGE-L score {rouge_l_score} is less than threshold {rouge_l_threshold}"
                    
                    print(f"Test passed for query: {case['query']}")
                    passed_tests += 1
                except AssertionError as e:
                    print(e)
                    print(f"Test failed for query: {case['query']}")
                    failed_tests += 1
            else:
                print(f"Test failed for query (no tool call): {case['query']}")
                failed_tests += 1
        else:
            # 验证没有调用工具
            if response.choices[0].message.tool_calls:
                print(f"Test failed for query (unexpected tool call): {case['query']}")
                failed_tests += 1
            else:
                print(f"Test passed for query: {case['query']}")
                passed_tests += 1

    total_tests = passed_tests + failed_tests
    accuracy = (passed_tests / total_tests) * 100 if total_tests > 0 else 0
    print(f"Total tests: {total_tests}, Passed: {passed_tests}, Failed: {failed_tests}, Accuracy: {accuracy:.2f}%")

def chat_completion_request(messages, tools=None, tool_choice=None, model=deployment_name, json_mode=False):
    try:
        params = {
            'model': model,
            'messages': messages,
            'tools': tools,
            'tool_choice': tool_choice
        }

        if json_mode:
            params['response_format'] = { "type": "json_object" }

        response = client.chat.completions.create(**params)
        return response

    except Exception as e:
        print("Unable to generate ChatCompletion response")
        print(f"Exception: {e}")
        return e

def evaluate_with_llm(predicted_arguments, expected_arguments, model, tools):
    prompt = f"""
    Compare the following predicted arguments and expected arguments for a function call.
    Predicted Arguments: {json.dumps(predicted_arguments, indent=2)}
    Expected Arguments: {json.dumps(expected_arguments, indent=2)}
    Evaluate if the predicted arguments sufficiently match the expected arguments, considering minor differences acceptable. Respond with a JSON object containing a single key "match" with a boolean value indicating if they match.

    Example response:
    {{
        "match": true
    }}
    """
    response = chat_completion_request(messages=[{"role": "system", "content": prompt}],model = model, json_mode=True)
    json_response = json.loads(response.choices[0].message.content)
    return json_response.get("match", False)

def run_tests_with_llm(test_cases, model=deployment_name, rouge_l_threshold=0.8):
    passed_tests = 0
    failed_tests = 0

    for case in test_cases:
        response = call_openai_tool(model, case['query'], tools)

        if case['expected_function'] and case['function_arguments']:
            if response.choices[0].message.tool_calls:
                tool_calls = response.choices[0].message.tool_calls

                # 验证函数调用名称和参数
                try:
                    assert tool_calls[0].function.name == case['expected_function'], f"Expected function {case['expected_function']}, but got {tool_calls[0].function.name}"
                    predicted_arguments = json.loads(tool_calls[0].function.arguments)
                    
                    # 使用 LLM 来判断参数匹配
                    match = evaluate_with_llm(predicted_arguments, case['function_arguments'], model, tools)
                    
                    assert match, "LLM evaluation determined the arguments do not match"
                    
                    print(f"Test passed for query: {case['query']}")
                    passed_tests += 1
                except AssertionError as e:
                    print(e)
                    print(f"Test failed for query: {case['query']}")
                    failed_tests += 1
            else:
                print(f"Test failed for query (no tool call): {case['query']}")
                failed_tests += 1
        else:
            # 验证没有调用工具
            if response.choices[0].message.tool_calls:
                print(f"Test failed for query (unexpected tool call): {case['query']}")
                failed_tests += 1
            else:
                print(f"Test passed for query: {case['query']}")
                passed_tests += 1

    total_tests = passed_tests + failed_tests
    accuracy = (passed_tests / total_tests) * 100 if total_tests > 0 else 0
    print(f"Total tests: {total_tests}, Passed: {passed_tests}, Failed: {failed_tests}, Accuracy: {accuracy:.2f}%")