Load artifacts from a directory
(self, artifact_dir: str)
| 2510 | logger.warning(f"Failed to write artifact {key} to {file_path}: {e}") |
| 2511 | |
| 2512 | def _load_artifact_dir(self, artifact_dir: str) -> Dict[str, Union[str, bytes]]: |
| 2513 | """Load artifacts from a directory""" |
| 2514 | artifacts = {} |
| 2515 | |
| 2516 | try: |
| 2517 | for filename in os.listdir(artifact_dir): |
| 2518 | file_path = os.path.join(artifact_dir, filename) |
| 2519 | if os.path.isfile(file_path): |
| 2520 | try: |
| 2521 | # Try to read as text first |
| 2522 | with open(file_path, "r", encoding="utf-8") as f: |
| 2523 | content = f.read() |
| 2524 | artifacts[filename] = content |
| 2525 | except UnicodeDecodeError: |
| 2526 | # If text fails, read as binary |
| 2527 | with open(file_path, "rb") as f: |
| 2528 | content = f.read() |
| 2529 | artifacts[filename] = content |
| 2530 | except Exception as e: |
| 2531 | logger.warning(f"Failed to read artifact file {file_path}: {e}") |
| 2532 | except Exception as e: |
| 2533 | logger.warning(f"Failed to list artifact directory {artifact_dir}: {e}") |
| 2534 | |
| 2535 | return artifacts |
| 2536 | |
| 2537 | def log_prompt( |
| 2538 | self, |