Merge pull request #986 from parea-ai/PAI-1321-trace-evals-python

feat: trace evals

parea/utils/trace_utils.py

@@ -6,6 +6,7 @@
 import logging
 import os
 import threading
+import traceback
 from collections import ChainMap
 from datetime import datetime
 from functools import wraps
@@ -43,7 +44,8 @@ def clear_trace_context():
 
 
 def log_in_thread(target_func: Callable, data: Dict[str, Any]):
-    logging_thread = threading.Thread(target=target_func, kwargs=data)
+    context = contextvars.copy_context()
+    logging_thread = threading.Thread(target=context.run, args=(target_func,), kwargs=data)
     logging_thread.start()
 
 
@@ -262,7 +264,7 @@ async def async_wrapper(*args, **kwargs):
                 return result
             except Exception as e:
                 logger.exception(f"Error occurred in function {func.__name__}, {e}")
-                fill_trace_data(trace_id, {"error": str(e)}, UpdateTraceScenario.ERROR)
+                fill_trace_data(trace_id, {"error": traceback.format_exc()}, UpdateTraceScenario.ERROR)
                 raise e
             finally:
                 try:
@@ -282,7 +284,7 @@ def wrapper(*args, **kwargs):
                 return result
             except Exception as e:
                 logger.exception(f"Error occurred in function {func.__name__}, {e}")
-                fill_trace_data(trace_id, {"error": str(e)}, UpdateTraceScenario.ERROR)
+                fill_trace_data(trace_id, {"error": traceback.format_exc()}, UpdateTraceScenario.ERROR)
                 raise e
             finally:
                 try:
@@ -305,7 +307,6 @@ def wrapper(*args, **kwargs):
 
 def call_eval_funcs_then_log(trace_id: str, eval_funcs: List[Callable] = None):
     data = trace_data.get()[trace_id]
-    # parea_logger.default_log(data=data)
 
     if eval_funcs and data.status == "success" and random() <= data.apply_eval_frac:
         thread_ids_running_evals.get().append(trace_id)
@@ -314,7 +315,7 @@ def call_eval_funcs_then_log(trace_id: str, eval_funcs: List[Callable] = None):
         scores = []
         for func in eval_funcs:
             try:
-                score = func(data)
+                score = trace()(func)(data)
                 if isinstance(score, EvaluationResult):
                     scores.append(score)
                 elif isinstance(score, list):
@@ -323,7 +324,6 @@ def call_eval_funcs_then_log(trace_id: str, eval_funcs: List[Callable] = None):
                     scores.append(EvaluationResult(name=func.__name__, score=score))
             except Exception as e:
                 logger.exception(f"Error occurred calling evaluation function '{func.__name__}', {e}", exc_info=e)
-        # parea_logger.update_log(data=UpdateLog(trace_id=trace_id, field_name_to_value_map={"scores": scores}))
         trace_data.get()[trace_id].scores = scores
         thread_ids_running_evals.get().remove(trace_id)
 

parea/wrapper/wrapper.py

@@ -130,7 +130,6 @@ def _init_trace(self, kwargs) -> Tuple[str, datetime, contextvars.Token]:
 
         return trace_id, start_time, token
 
-    @skip_decorator_if_func_in_stack(call_eval_funcs_then_log, _make_evaluations)
     def async_decorator(self, orig_func: Callable) -> Callable:
         @functools.wraps(orig_func)
         async def wrapper(*args, **kwargs):
@@ -162,7 +161,6 @@ async def wrapper(*args, **kwargs):
 
         return wrapper
 
-    @skip_decorator_if_func_in_stack(call_eval_funcs_then_log, _make_evaluations)
     def sync_decorator(self, orig_func: Callable) -> Callable:
         @functools.wraps(orig_func)
         def wrapper(*args, **kwargs):

pyproject.toml

@@ -6,7 +6,7 @@ build-backend = "poetry.core.masonry.api"
 [tool.poetry]
 name = "parea-ai"
 packages = [{ include = "parea" }]
-version = "0.2.182"
+version = "0.2.183"
 description = "Parea python sdk"
 readme = "README.md"
 authors = ["joel-parea-ai <[email protected]>"]