Merge pull request #922 from parea-ai/fix-lamda-invocation

helper for Lambdas

parea/__init__.py

@@ -18,7 +18,7 @@
 from parea.experiment.experiment import Experiment
 from parea.helpers import gen_trace_id, write_trace_logs_to_csv
 from parea.parea_logger import parea_logger
-from parea.utils.trace_utils import get_current_trace_id, get_root_trace_id, trace, trace_insert
+from parea.utils.trace_utils import clear_trace_context, get_current_trace_id, get_root_trace_id, trace, trace_insert
 from parea.wrapper.openai_raw_api_tracer import aprocess_stream_and_yield, process_stream_and_yield
 from parea.wrapper.utils import convert_openai_raw_to_log
 

parea/utils/trace_integrations/langchain.py

@@ -1,5 +1,6 @@
 from typing import Union
 
+import logging
 from uuid import UUID
 
 from langchain_core.tracers import BaseTracer
@@ -11,6 +12,8 @@
 from parea.schemas.log import TraceIntegrations
 from parea.utils.trace_utils import fill_trace_data, get_current_trace_id, get_root_trace_id
 
+logger = logging.getLogger()
+
 
 class PareaAILangchainTracer(BaseTracer):
     parent_trace_id: UUID
@@ -24,16 +27,17 @@ def _persist_run(self, run: Union[Run, LLMRun, ChainRun, ToolRun]) -> None:
         # using .dict() since langchain Run class currently set to Pydantic v1
         data = run.dict()
         data["_parea_root_trace_id"] = self._parea_root_trace_id or None
-        if run.execution_order == 1:
+        # check if run has an attribute execution order
+        if hasattr(run, "execution_order") and run.execution_order == 1:
             data["_parea_parent_trace_id"] = self._parea_parent_trace_id or None
         parea_logger.record_vendor_log(data, TraceIntegrations.LANGCHAIN)
 
     def get_parent_trace_id(self) -> UUID:
         return self.parent_trace_id
 
     def _on_run_create(self, run: Run) -> None:
-        if run.execution_order == 1:
-            # need to check if any traces already exist\
+        if hasattr(run, "execution_order") and run.execution_order == 1:
+            # need to check if any traces already exist
             self._parea_root_trace_id = get_root_trace_id()
             if parent_trace_id := get_current_trace_id():
                 self._parea_parent_trace_id = parent_trace_id

parea/utils/trace_utils.py

@@ -34,6 +34,14 @@
 execution_order_counters = contextvars.ContextVar("execution_order_counters", default={})
 
 
+def clear_trace_context():
+    global trace_context, trace_data, thread_ids_running_evals, execution_order_counters
+    trace_context = contextvars.ContextVar("trace_context", default=[])
+    trace_data = contextvars.ContextVar("trace_data", default={})
+    thread_ids_running_evals = contextvars.ContextVar("thread_ids_running_evals", default=[])
+    execution_order_counters = contextvars.ContextVar("execution_order_counters", default={})
+
+
 def log_in_thread(target_func: Callable, data: Dict[str, Any]):
     logging_thread = threading.Thread(target=target_func, kwargs=data)
     logging_thread.start()

pyproject.toml

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