Skip to content

Class swarmauri_standard.tracing.ChainTracer.ChainTracer

swarmauri_standard.tracing.ChainTracer.ChainTracer 

ChainTracer()

Bases: IChainTracer

Source code in swarmauri_standard/tracing/ChainTracer.py 
6
7
def __init__(self):
    self.traces = []

traces instance-attribute 

traces = []

process_chain 

process_chain(chain)

Processes each item in the operation chain by executing the specified external function with its args and kwargs. Logs starting, annotating, and ending the trace based on tuple position.

PARAMETER DESCRIPTION
chain

A list where each tuple contains: - An external function to execute. - A list of positional arguments for the function. - A dictionary of keyword arguments for the function.

TYPE: List[Tuple[Callable[..., Any], List[Any], Dict[str, Any]]]

Source code in swarmauri_standard/tracing/ChainTracer.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
def process_chain(
    self, chain: List[Tuple[Callable[..., Any], List[Any], Dict[str, Any]]]
) -> "ChainTracer":
    """
    Processes each item in the operation chain by executing the specified external function
    with its args and kwargs. Logs starting, annotating, and ending the trace based on tuple position.

    Args:
        chain (List[Tuple[Callable[..., Any], List[Any], Dict[str, Any]]]): A list where each tuple contains:
            - An external function to execute.
            - A list of positional arguments for the function.
            - A dictionary of keyword arguments for the function.
    """
    for i, (func, args, kwargs) in enumerate(chain):
        # Infer operation type and log

        if i == 0:
            operation = "Start"
            self.start_trace(*args, **kwargs)
        elif i == len(chain) - 1:
            operation = "End"
            self.end_trace(*args, **kwargs)
        else:
            operation = "Annotate"
            self.annotate_trace(*args, **kwargs)

        # For the actual external function call
        result = func(*args, **kwargs)
        print(f"Function '{func.__name__}' executed with result: {result}")

        self.traces.append((operation, func, args, kwargs, result))

    return self

start_trace 

start_trace(*args, **kwargs)
Source code in swarmauri_standard/tracing/ChainTracer.py 
43
44
def start_trace(self, *args, **kwargs) -> None:
    print(f"Starting trace with args: {args}, kwargs: {kwargs}")

annotate_trace 

annotate_trace(*args, **kwargs)
Source code in swarmauri_standard/tracing/ChainTracer.py 
46
47
def annotate_trace(self, *args, **kwargs) -> None:
    print(f"Annotating trace with args: {args}, kwargs: {kwargs}")

end_trace 

end_trace(*args, **kwargs)
Source code in swarmauri_standard/tracing/ChainTracer.py 
49
50
def end_trace(self, *args, **kwargs) -> None:
    print(f"Ending trace with args: {args}, kwargs: {kwargs}")

show 

show()
Source code in swarmauri_standard/tracing/ChainTracer.py 
52
53
54
def show(self) -> None:
    for entry in self.traces:
        print(entry)