Berack/upo-app-agents
0
0
Fork 0
Code Issues Pull Requests 3 Actions Packages Projects Releases Wiki Activity

Enhance error logging in WrapperHandler to provide concise traceback information

Browse Source
This commit is contained in:
Giacomo Bertolazzi
2025-10-01 12:49:43 +02:00
parent 73dcbbe12b
commit 2aa97e264a
1 changed files with 7 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
src/app/utils/wrapper_handler.py
View File

@@ -1,4 +1,5 @@
import time import time
import traceback
from typing import TypeVar, Callable, Generic, Iterable, Type from typing import TypeVar, Callable, Generic, Iterable, Type
from agno.utils.log import log_warning, log_info from agno.utils.log import log_warning, log_info
@@ -54,7 +55,7 @@ class WrapperHandler(Generic[W]):
return result return result
except Exception as e: except Exception as e:
self.retry_count += 1 self.retry_count += 1
log_warning(f"{wrapper} failed {self.retry_count}/{self.retry_per_wrapper}: {e}") log_warning(f"{wrapper} failed {self.retry_count}/{self.retry_per_wrapper}: {WrapperHandler.__concise_error(e)}")
if self.retry_count >= self.retry_per_wrapper: if self.retry_count >= self.retry_per_wrapper:
self.index = (self.index + 1) % len(self.wrappers) self.index = (self.index + 1) % len(self.wrappers)
@@ -84,7 +85,7 @@ class WrapperHandler(Generic[W]):
result = func(wrapper) result = func(wrapper)
results[wrapper.__class__] = result results[wrapper.__class__] = result
except Exception as e: except Exception as e:
log_warning(f"{wrapper} failed: {e}") log_warning(f"{wrapper} failed: {WrapperHandler.__concise_error(e)}")
if not results: if not results:
raise Exception("All wrappers failed") raise Exception("All wrappers failed")
return results return results
@@ -93,6 +94,10 @@ class WrapperHandler(Generic[W]):
def __check(wrappers: list[W]) -> bool: def __check(wrappers: list[W]) -> bool:
return all(w.__class__ is type for w in wrappers) return all(w.__class__ is type for w in wrappers)
def __concise_error(e: Exception) -> str:
last_frame = traceback.extract_tb(e.__traceback__)[-1]
return f"{e} [\"{last_frame.filename}\", line {last_frame.lineno}]"
@staticmethod @staticmethod
def build_wrappers(constructors: Iterable[Type[W]], try_per_wrapper: int = 3, retry_delay: int = 2) -> 'WrapperHandler[W]': def build_wrappers(constructors: Iterable[Type[W]], try_per_wrapper: int = 3, retry_delay: int = 2) -> 'WrapperHandler[W]':
""" """