printer

TraceHTMLPrinter

TraceHTMLPrinter()

Bases: TracePrinterBase

The printer used to print the trace in the format of HTML.

Source code in src/appl/tracing/printer.py

def __init__(self):
    """Initialize the printer."""
    super().__init__()
    self._generation_style = "text-success-emphasis bg-Success-subtle list-group-item d-flex justify-content-between align-items-center"
    self._time_style = "position-absolute top-0 start-100 translate-middle badge rounded-pill bg-info"
    self._cost_style = "position-absolute bottom-0 start-100 translate-middle badge rounded-pill bg-warning"
    self._longest_shown_output = 70

    self._head = load_file(os.path.join(folder, "header.html"))
    self._color_map = {
        "user": "text-bg-info",
        "assistant": "text-bg-warning",
        "system": "text-bg-success",
    }

print

print(
    trace: TraceEngineBase,
    trace_metadata: Optional[Dict[str, Any]] = None,
) -> str

Print the trace in the format of HTML.

Source code in src/appl/tracing/printer.py

@ppl
def print(
    self, trace: TraceEngineBase, trace_metadata: Optional[Dict[str, Any]] = None
) -> str:
    """Print the trace in the format of HTML."""
    with Tagged("html"):
        self._head
        with Tagged("body"):
            for node in trace.trace_nodes.values():
                if node.parent is None:
                    self._print_node(node, trace.min_timestamp)
    if trace_metadata:
        with Tagged("table", attrs={"class": "table small"}):
            if start_time := trace_metadata.get("start_time", None):
                self._make_line("Start Time", start_time)
            self._make_line(
                "Full Configs", f"<pre>{yaml.dump(trace_metadata)}</pre>"
            )
    return str(records())

TraceLangfusePrinter

Bases: TracePrinterBase

The printer used to print the trace to langfuse.

print

print(
    trace: TraceEngineBase,
    trace_metadata: Optional[Dict[str, Any]] = None,
) -> None

Print the trace to langfuse.

Source code in src/appl/tracing/printer.py

def print(
    self, trace: TraceEngineBase, trace_metadata: Optional[Dict[str, Any]] = None
) -> None:
    """Print the trace to langfuse."""
    from langfuse import Langfuse
    from langfuse.client import StatefulTraceClient

    project_public_key = os.environ.get("LANGFUSE_PUBLIC_KEY", None)
    if project_public_key is None:
        raise ValueError("LANGFUSE_PUBLIC_KEY is not set")
    url = os.environ.get("LANGFUSE_HOST", "http://localhost:3000")
    logger.info(f"project_public_key: {project_public_key}, api url: {url}")

    metadata = trace_metadata or {}
    user_id = metadata.get("user_id", None)
    if user_id is None:
        git_info: Dict[str, Any] = metadata.get("git_info", {})
        user_id = git_info.get("git_user_email", "unknown")

    session_id = metadata.get("start_time", str(uuid.uuid4()))
    if "exec_file_basename" in metadata:
        base_name = metadata["exec_file_basename"]
        # Use basename + start_time as the session_id
        session_id = f"[{base_name}] {session_id}"
    else:
        base_name = metadata.get("name", "main")

    root = Langfuse().trace(
        name=base_name,
        user_id=user_id,
        session_id=session_id,
        metadata=metadata,
    )

    def visit_tree(node: TraceNode, trace_node: StatefulTraceClient) -> None:
        """Visit the trace node tree and send the trace event to langfuse."""
        start_time = datetime.fromtimestamp(node.start_time, timezone.utc)
        if node.end_time == 0.0:
            logger.warning(f"trace event {node.name} does not finish")
            end_time = start_time
        else:
            end_time = datetime.fromtimestamp(node.end_time, timezone.utc)

        logger.info(f"sending trace event {node.name} with type {node.type}")
        if node.type == "func":
            metadata = node.metadata or {}
            # metadata = metadata.copy()
            # if "source_code" in metadata:
            #     metadata["source_code"] = (
            #         "\n```python\n" + metadata["source_code"] + "\n```\n"
            #     )
            client = trace_node.span(
                name=node.name,
                start_time=start_time,
                end_time=end_time,
                input=node.args,
                output=node.ret,
                metadata=metadata,
            )
        elif node.type in ["gen", "raw_llm"]:
            inputs = (node.args or {}).copy()
            outputs = node.ret

            model_name = inputs.pop("model", None)
            metadata = inputs.pop("metadata", {})
            metadata = (node.metadata or {}) | metadata

            def extra_supported_model_parameters(inputs: Dict) -> Dict:
                parameters = {}
                for k, v in inputs.items():
                    if k not in [
                        "messages",
                        "tools",
                        "response_format",
                        "tool_choice",
                    ]:
                        parameters[k] = v

                if tool_choice := inputs.get("tool_choice", None):
                    if isinstance(tool_choice, str):
                        name = tool_choice
                    elif isinstance(tool_choice, dict):
                        name = tool_choice.get("function", {}).get("name", None)
                    else:
                        logger.warning(f"unknown tool_choice: {tool_choice}")

                    if name:
                        parameters["tool_choice"] = name

                for k in parameters.keys():
                    inputs.pop(k, None)

                return parameters

            model_parameters = extra_supported_model_parameters(inputs)

            usage = None
            if outputs is not None:
                if node.type == "gen":
                    model_name = None
                elif not isinstance(outputs, str):
                    outputs: ModelResponse = outputs  # type: ignore
                    usage = outputs.usage
                    message = outputs.choices[0].message  # type: ignore
                    outputs = message.content
                    if message.tool_calls:
                        outputs = {
                            "content": message.content,
                            "tool_calls": [
                                f"ToolCall(id={tool.id}, name={tool.function.name}, args={tool.function.arguments})"
                                for tool in message.tool_calls
                            ],
                        }
            else:
                outputs = "Not Finished."

            client = trace_node.generation(
                name=node.name,
                start_time=start_time,
                end_time=end_time,
                input=inputs,
                output=outputs,
                model=model_name,
                model_parameters=model_parameters,
                usage=usage,
                metadata=metadata,
            )
        else:
            raise ValueError(f"Unknown node type: {node.type}")
        for child in node.children:
            visit_tree(child, client)

    for node in trace.trace_nodes.values():
        if node.parent is None:
            visit_tree(node, root)

TraceLunaryPrinter

Bases: TracePrinterBase

The printer used to log the trace to lunary.

print

print(
    trace: TraceEngineBase,
    trace_metadata: Optional[Dict[str, Any]] = None,
) -> None

Log the trace to lunary.

Source code in src/appl/tracing/printer.py

def print(
    self, trace: TraceEngineBase, trace_metadata: Optional[Dict[str, Any]] = None
) -> None:
    """Log the trace to lunary."""
    import lunary

    project_id = os.environ.get("LUNARY_PUBLIC_KEY", None)
    if project_id is None:
        raise ValueError("LUNARY_PUBLIC_KEY is not set")
    url = os.environ.get("LUNARY_API_URL", "http://localhost:3333")
    logger.info(f"project_id: {project_id}, api url: {url}")
    lunary.config(app_id=project_id, api_url=url)

    suffix = f"_{uuid.uuid4().hex}"
    logger.info(f"suffix: {suffix}")

    user_id = None
    if trace_metadata:
        user_id = trace_metadata.get("user_id", None)

    def get_parent_run_id(node: TraceNode) -> Optional[str]:
        if node.parent is None:
            return None
        return node.parent.name + suffix

    """Log the trace to lunary."""
    for node in trace.trace_nodes.values():
        if node.type == "func":
            logger.info(
                f"sending func event {node.name} to lunary with parent {get_parent_run_id(node)}"
            )
            lunary.track_event(
                "chain",
                "start",
                run_id=node.name + suffix,
                name=node.name,
                parent_run_id=get_parent_run_id(node),
                input=node.args,
                timestamp=timestamp_to_iso(node.start_time),
                user_id=user_id,
            )
            lunary.track_event(
                "chain",
                "end",
                run_id=node.name + suffix,
                output=node.ret,
                timestamp=timestamp_to_iso(node.end_time),
            )

        elif node.type in ["gen", "raw_llm"]:
            merged_metadata: Dict[str, Any] = node.metadata or {}
            extra_info = copy.deepcopy(node.args or {})
            model_name = extra_info.pop("model", node.name)
            messages = extra_info.pop("messages", "")
            merged_metadata.update(extra_info)

            if node.type == "gen":
                logger.info(
                    f"sending llm event {node.name} to lunary with parent {get_parent_run_id(node)}"
                )

                # skip the raw generation, support for legacy traces
                # if node.name.endswith("_raw"):
                #     continue
                merged_metadata["gen_ID"] = node.name
                lunary.track_event(
                    "llm",
                    "start",
                    run_id=node.name + suffix,
                    name=model_name,
                    parent_run_id=get_parent_run_id(node),
                    metadata=merged_metadata,
                    input=messages,
                    timestamp=timestamp_to_iso(node.start_time),
                    user_id=user_id,
                )
                lunary.track_event(
                    "llm",
                    "end",
                    run_id=node.name + suffix,
                    output={"role": "assistant", "content": node.ret},
                    timestamp=timestamp_to_iso(node.end_time),
                )
            elif node.type == "raw_llm":
                logger.info(
                    f"sending raw llm event {node.name} to lunary with parent {get_parent_run_id(node)}"
                )
                lunary.track_event(
                    "llm",
                    "start",
                    run_id=node.name + suffix,
                    name=model_name,
                    parent_run_id=get_parent_run_id(node),
                    metadata=merged_metadata,
                    input=messages,
                    timestamp=timestamp_to_iso(node.start_time),
                    user_id=user_id,
                )
                response: ModelResponse = node.ret  # complete response
                message = response.choices[0].message  # type: ignore
                output = {
                    "role": "assistant",
                    "content": message.content or "",  # type: ignore
                }
                if message.tool_calls:
                    output["tool_calls"] = [
                        {
                            "id": tool.id,
                            "type": "function",
                            "function": {
                                "name": tool.function.name,
                                "arguments": tool.function.arguments,
                            },
                        }
                        for tool in message.tool_calls
                        # TODO: support tool calls
                    ]
                lunary.track_event(
                    "llm",
                    "end",
                    run_id=node.name + suffix,
                    output=output,
                    timestamp=timestamp_to_iso(node.end_time),
                )

TraceProfilePrinter

TraceProfilePrinter(display_functions: bool = False)

Bases: TracePrinterBase

The printer used to print the trace in the format of profile.

Parameters:

• display_functions (bool, default: False ) –

  Whether to display the function calls.

Source code in src/appl/tracing/printer.py

def __init__(self, display_functions: bool = False):
    """Initialize the printer.

    Args:
        display_functions: Whether to display the function calls.
    """
    self._display_functions = display_functions

build_event

build_event(
    event: TraceEventBase, min_timestamp: float
) -> Dict

Build the event for the trace.

Source code in src/appl/tracing/printer.py

def build_event(self, event: TraceEventBase, min_timestamp: float) -> Dict:
    """Build the event for the trace."""
    ts = str((event.time_stamp - min_timestamp) * 1e6)
    data = {"pid": 0, "tid": 0, "name": event.name, "ts": ts}
    # TODO: add args to the trace
    if isinstance(event, CompletionRequestEvent):
        data["cat"] = "gen"
        data["ph"] = "b"
        data["id"] = event.name
    elif isinstance(event, CompletionResponseEvent):
        data["cat"] = "gen"
        data["ph"] = "e"
        data["id"] = event.name
        data["cost"] = event.cost
        data["output"] = event.ret.dict()
    elif self._display_functions:
        if isinstance(event, FunctionCallEvent):
            data["cat"] = "func"
            data["ph"] = "B"
            data["tid"] = "main"
        elif isinstance(event, FunctionReturnEvent):
            data["cat"] = "func"
            data["ph"] = "E"
            data["tid"] = "main"
    return data

print

print(
    trace: TraceEngineBase,
    trace_metadata: Optional[Dict[str, Any]] = None,
) -> Dict

Print the trace in the format of Chrome tracing.

Source code in src/appl/tracing/printer.py

def print(
    self, trace: TraceEngineBase, trace_metadata: Optional[Dict[str, Any]] = None
) -> Dict:
    """Print the trace in the format of Chrome tracing."""
    events = []
    for event in trace.events:
        if data := self.build_event(event, trace.min_timestamp):
            events.append(data)
    return {"traceEvents": events}

TraceYAMLPrinter

Bases: TracePrinterBase

The printer used to print the trace in the format of YAML.

print

print(
    trace: TraceEngineBase,
    trace_metadata: Optional[Dict[str, Any]] = None,
) -> None

Print the trace in the format of YAML.

Source code in src/appl/tracing/printer.py

def print(
    self, trace: TraceEngineBase, trace_metadata: Optional[Dict[str, Any]] = None
) -> None:
    """Print the trace in the format of YAML."""
    # TODO: implement the YAML printer
    pass

print_trace

print_trace(
    printer: Optional[TracePrinterBase] = None,
    trace_file: Optional[str] = None,
) -> None

Print to visualize the trace.

Default printer is to the langfuse platform. You can also configured to the local hosted version.

Source code in src/appl/tracing/printer.py

def print_trace(
    printer: Optional[TracePrinterBase] = None,
    trace_file: Optional[str] = None,
) -> None:
    """Print to visualize the trace.

    Default printer is to the [langfuse](https://langfuse.com/) platform.
    You can also configured to the local hosted version.
    """
    if printer is None:
        printer = TraceLangfusePrinter()
    if trace_file is None:
        trace = global_vars.trace_engine
        if trace is None:
            raise ValueError("No trace found")
        trace_file = global_vars.metadata.trace_file
    else:
        trace = TraceEngine(trace_file)

    if trace_file is not None:
        meta_file = get_meta_file(trace_file)
        trace_metadata = load_file(meta_file)
    else:
        trace_metadata = None
    printer.print(trace, trace_metadata)

timestamp_to_iso

timestamp_to_iso(time_stamp: float) -> str

Convert the timestamp to the ISO format.

Source code in src/appl/tracing/printer.py

def timestamp_to_iso(time_stamp: float) -> str:
    """Convert the timestamp to the ISO format."""
    return datetime.fromtimestamp(time_stamp, timezone.utc).isoformat()