8
8
from functools import wraps
9
9
from typing import Any , Dict , Generator , List , Optional , Tuple
10
10
11
+ from .. import utils
11
12
from ..services import data_streamer
12
13
from . import enums , steps , traces
13
14
14
15
logger = logging .getLogger (__name__ )
15
16
17
+ _publish = utils .get_env_variable ("PUBLISH" ) == "true"
16
18
_streamer = None
17
- try :
18
- _streamer = data_streamer .DataStreamer (publish = True )
19
- # pylint: disable=broad-except
20
- except Exception as exc :
21
- logger .error (
22
- "You have not provided enough information to upload traces to Openlayer."
23
- "\n %s \n "
24
- "To upload the traces, please provide the missing information and try again." ,
25
- exc ,
26
- )
19
+ if _publish :
20
+ _streamer = data_streamer .DataStreamer ()
27
21
28
22
_current_step = contextvars .ContextVar ("current_step" )
29
23
_current_trace = contextvars .ContextVar ("current_trace" )
30
24
31
25
26
+ def get_current_trace () -> Optional [traces .Trace ]:
27
+ """Returns the current trace."""
28
+ return _current_trace .get (None )
29
+
30
+
31
+ def get_current_step () -> Optional [steps .Step ]:
32
+ """Returns the current step."""
33
+ return _current_step .get (None )
34
+
35
+
32
36
@contextmanager
33
37
def create_step (
34
38
name : str ,
@@ -43,7 +47,7 @@ def create_step(
43
47
)
44
48
new_step .start_time = time .time ()
45
49
46
- parent_step : Optional [steps .Step ] = _current_step . get ( None )
50
+ parent_step : Optional [steps .Step ] = get_current_step ( )
47
51
is_root_step : bool = parent_step is None
48
52
49
53
if parent_step is None :
@@ -53,7 +57,7 @@ def create_step(
53
57
current_trace .add_step (new_step )
54
58
else :
55
59
logger .debug ("Adding step %s to parent step %s" , name , parent_step .name )
56
- current_trace = _current_trace . get ()
60
+ current_trace = get_current_trace ()
57
61
parent_step .add_nested_step (new_step )
58
62
59
63
token = _current_step .set (new_step )
@@ -86,14 +90,11 @@ def create_step(
86
90
"prompt" : new_step .inputs .get ("prompt" ),
87
91
}
88
92
)
89
- if _streamer :
90
- _streamer .stream_data (data = trace_data , config = config )
91
- else :
92
- logger .warning (
93
- "Trace computed but not uploaded to Openlayer. "
94
- "You have not provided enough information to upload traces to"
95
- " Openlayer."
96
- )
93
+ if _publish :
94
+ try :
95
+ _streamer .stream_data (data = trace_data , config = config )
96
+ except Exception as _ :
97
+ logger .error ("Could not stream data to Openlayer" )
97
98
else :
98
99
logger .debug ("Ending step %s" , name )
99
100
0 commit comments