File size: 10,428 Bytes
cfcbbc8 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 |
import os
import pandas as pd
import re
import glob
from tqdm import tqdm
import datetime
import openai
import argparse
import io
def summarize_results(results_dirs, output_csv, model, no_llm = False):
client = openai.OpenAI(
api_key = os.environ.get('CBORG_API_KEY'),
base_url = 'https://api.cborg.lbl.gov'
)
error_categorization_prompt = (
"You are an expert at classifying error messages from machine learning workflows in high energy physics.\n\n"
"Workflow summary:\n"
"- A user provides an analysis task prompt.\n"
"- A supervisor agent breaks down the task and instructs a coder agent.\n"
"- The coder agent generates code, which is executed.\n"
"- The supervisor reviews results and may iterate with the coder to fix issues until the task is complete.\n"
"Below is a list of error categories:\n"
"all data weights = 0, "
"dummy data created, "
"function-calling error, "
"incorrect branch name, "
"intermediate file not found, "
"semantic error, "
"other."
"Your task: For the given error description, select the single most appropriate error category from the list above. "
"Base your choice on the underlying nature or root cause of the error, not on the symptoms, error messages, or observable effects. "
"Focus on what fundamentally caused the error, such as logic mistakes, missing dependencies, data mismatches, or miscommunication, rather than how the error was reported or observed.\n"
"Return ALL applicable category names, each wrapped with three asterisks on each side, separated by commas, like this: ***Category***"
"Do not include any other text, explanation, or formatting."
"log file:\n"
)
results = []
for results_dir in results_dirs:
for name in tqdm(os.listdir(results_dir), desc=f"generating error descriptions for {results_dir}"):
output_dir = os.path.join(results_dir, name)
if os.path.isdir(output_dir):
# Extract config (everything before "_step")
config_match = re.match(r'^(.*?)_step\d+', name)
config = config_match.group(1) if config_match else None
# Extract step (int after "_step")
step_match = re.search(r'_step(\d+)', name)
step = int(step_match.group(1)) if step_match else None
result = {
"supervisor": None,
"coder": None,
"step": step,
"success": None,
"iterations": None,
"duration": None,
"API_calls": None,
"input_tokens": None,
"output_tokens": None,
"user_prompt_tokens": None,
"supervisor_to_coder_tokens": None,
"coder_output_tokens": None,
"feedback_to_supervisor_tokens": None,
"error": "Uncategorized",
"error_description": None,
"output_dir": output_dir,
}
log_dir = os.path.join(output_dir, "logs")
if os.path.isdir(log_dir):
comp_log_files = glob.glob(os.path.join(log_dir, "*comprehensive_log.txt"))
comp_log_str = None
if comp_log_files:
with open(comp_log_files[0], "r") as f:
comp_log_str = f.read()
else:
result["success"] = False
result["error_description"] = "comprehensive log file not found"
results.append(result)
continue
supervisor_match = re.search(r"Supervisor:\s*([^\s]+)", comp_log_str)
coder_match = re.search(r"Coder:\s*([^\s]+)", comp_log_str)
if supervisor_match:
result["supervisor"] = supervisor_match.group(1)
if coder_match:
result["coder"] = coder_match.group(1)
iterations_match = re.search(r"Total Iterations:\s*(\d+)", comp_log_str)
if iterations_match:
result["iterations"] = int(iterations_match.group(1))
duration_match = re.search(r"Duration:\s*([0-9:.\s]+)", comp_log_str)
if duration_match:
duration_str = duration_match.group(1).strip()
try:
t = datetime.datetime.strptime(duration_str, "%H:%M:%S.%f")
except ValueError:
t = datetime.datetime.strptime(duration_str, "%H:%M:%S")
result["duration"] = t.hour * 3600 + t.minute * 60 + t.second + t.microsecond / 1e6
api_calls_match = re.search(r"Total API Calls:\s*(\d+)", comp_log_str)
if api_calls_match:
result["API_calls"] = int(api_calls_match.group(1))
input_tokens_match = re.search(r"Total Input Tokens:\s*(\d+)", comp_log_str)
if input_tokens_match:
result["input_tokens"] = int(input_tokens_match.group(1))
output_tokens_match = re.search(r"Total Output Tokens:\s*(\d+)", comp_log_str)
if output_tokens_match:
result["output_tokens"] = int(output_tokens_match.group(1))
match = re.search(r"User Prompt Tokens:\s*(\d+)", comp_log_str)
if match:
result["user_prompt_tokens"] = int(match.group(1))
match = re.search(r"Supervisor to Coder Tokens:\s*(\d+)", comp_log_str)
if match:
result["supervisor_to_coder_tokens"] = int(match.group(1))
match = re.search(r"Coder Output Tokens:\s*(\d+)", comp_log_str)
if match:
result["coder_output_tokens"] = int(match.group(1))
match = re.search(r"Feedback to Supervisor Tokens:\s*(\d+)", comp_log_str)
if match:
result["feedback_to_supervisor_tokens"] = int(match.group(1))
# Check validation.log to see if outputs are correct
val_log_files = glob.glob(os.path.join(log_dir, "*validation.log"))
val_log_str = None
if val_log_files:
with open(val_log_files[0], "r") as f:
val_log_str = f.read()
matches = re.findall(r'(✅ Validation successful|❌ Validation failed)', val_log_str)
if not matches:
result["success"] = False
else:
last = matches[-1]
result["success"] = last == "✅ Validation successful"
if (no_llm):
if (result["success"]):
result["error"] = None
else:
result["error"] = "Validation Error"
val_log_str = val_log_str.replace('\n', '').replace('\r', '')
else:
result["success"] = False
val_log_str = ""
if (not no_llm):
try:
response = client.chat.completions.create(
model = model,
messages = [
{
'role': 'user',
'content': error_categorization_prompt +
"\nComprehensive Log:\n" + comp_log_str +
"\nValidation Log:\n" + val_log_str
}
],
)
error_description = response.choices[-1].message.content
def parse_categories(llm_output):
# Find all ***Category Name*** matches
return [cat.strip() for cat in re.findall(r"\*\*\*(.*?)\*\*\*", llm_output)]
result["Error"] = parse_categories(error_description)
except Exception as e:
result["Error"] = "uncategorized"
print(error_description)
exit()
print(f"OpenAI API error: {e}")
else:
if ("API call failed" in comp_log_str):
result["error"] = "API Call Error"
else:
result["success"] = False
result["Error"] = "job submission failure"
results.append(result)
df = pd.DataFrame(results)
df = df.sort_values(by=["supervisor", "coder", "step", "output_dir"])
df.to_csv(output_csv, index=False)
print(f"Results written to {output_csv}")
def main():
parser = argparse.ArgumentParser(description="Summarize experiment logs and errors")
parser.add_argument("--results_dir", type=str, default=" ", nargs='+', required=False, help="One or more directories containing experiment results")
parser.add_argument("--output_csv", type=str, default="results_summary.csv", help="Path to output CSV file")
parser.add_argument("--model", type=str, default="gpt-oss-120b", help="LLM model to use for error summarization")
parser.add_argument("--no_llm", action="store_true", default=False, help="If set, only generate the CSV without LLM error description or categorization")
args = parser.parse_args()
summarize_results(
results_dirs=args.results_dir,
output_csv=args.output_csv,
model=args.model,
no_llm=args.no_llm
)
if __name__ == "__main__":
main() |