Function single_round_statistics

WebAgent/WebResummer/src/evaluate.py:60–129

Calculate statistics for a single round

(input_file, available_tools=None)

def single_round_statistics(input_file, available_tools=None):  
    """Calculate statistics for a single round"""
    def avg_statistic(value_list):
        if value_list:
            return sum(value_list) / len(value_list)
        return 0 
    
    try:
        with open(input_file, 'r', encoding='utf-8') as f: 
            samples = [json.loads(line) for line in f]
    except Exception as e:
        print(f"Error loading file {input_file}: {e}")
        return {}

    num_invalid = 0 
    tool_invocation = defaultdict(list)
    answer_lengths, traj_lengths = [], []
    
    try:
        tokenizer = AutoTokenizer.from_pretrained("/path/to/your/Qwen2.5-72B-Instruct")
    except Exception as e: 
        import tiktoken
        tokenizer = tiktoken.encoding_for_model("gpt-4o")
    
    for sample in samples:
        msgs = sample.get("messages", [])
        final_msg = msgs[-1]["content"] if len(msgs) else "" 

        if "<answer>" not in final_msg or "</answer>" not in final_msg:  
            num_invalid += 1 
            answer_length = 0 
        else:
            answer = final_msg.split("<answer>")[1].split("</answer>")[0].strip()
            answer_length = len(tokenizer.encode(answer))
        answer_lengths.append(answer_length)
        
        cur_tool_invocation = defaultdict(int)
        for msg in msgs:
            if msg["role"] == "assistant": 
                try:
                    tool_call = msg["content"].split("<tool_call>")[1].split("</tool_call>")[0].strip() 
                    tool_call = json.loads(tool_call) 
                    tool_name = tool_call["name"] 
                    if available_tools and tool_name in available_tools:
                        cur_tool_invocation[tool_name] += 1 
                    else:
                        cur_tool_invocation["invalid"] += 1 
                    cur_tool_invocation["total"] += 1 
                except:
                    continue 
        
        for k, v in cur_tool_invocation.items(): 
            tool_invocation[k].append(v)
        
        traj_length = len(tokenizer.encode("".join(msg["content"] for msg in msgs)))
        traj_lengths.append(traj_length)
    
    metrics = {