Skip to content

Llama debug

LlamaDebugHandler #

Bases: PythonicallyPrintingBaseHandler

回调处理程序，用于跟踪调试信息。

注意：这是一个测试版功能。在我们的代码库中使用和接口可能会发生变化。

该处理程序简单地跟踪事件的开始/结束，根据事件类型进行分隔。 您可以使用此回调处理程序来跟踪和调试事件。

Source code in llama_index/core/callbacks/llama_debug.py 
 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
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
class LlamaDebugHandler(PythonicallyPrintingBaseHandler):
    """回调处理程序，用于跟踪调试信息。

    注意：这是一个测试版功能。在我们的代码库中使用和接口可能会发生变化。

    该处理程序简单地跟踪事件的开始/结束，根据事件类型进行分隔。
    您可以使用此回调处理程序来跟踪和调试事件。

    Args:
        event_starts_to_ignore（可选[List[CBEventType]]）：要忽略跟踪事件开始时的事件类型列表。
        event_ends_to_ignore（可选[List[CBEventType]]）：要忽略跟踪事件结束时的事件类型列表。"""

    def __init__(
        self,
        event_starts_to_ignore: Optional[List[CBEventType]] = None,
        event_ends_to_ignore: Optional[List[CBEventType]] = None,
        print_trace_on_end: bool = True,
        logger: Optional[logging.Logger] = None,
    ) -> None:
        """初始化羊驼调试处理程序。"""
        self._event_pairs_by_type: Dict[CBEventType, List[CBEvent]] = defaultdict(list)
        self._event_pairs_by_id: Dict[str, List[CBEvent]] = defaultdict(list)
        self._sequential_events: List[CBEvent] = []
        self._cur_trace_id: Optional[str] = None
        self._trace_map: Dict[str, List[str]] = defaultdict(list)
        self.print_trace_on_end = print_trace_on_end
        event_starts_to_ignore = (
            event_starts_to_ignore if event_starts_to_ignore else []
        )
        event_ends_to_ignore = event_ends_to_ignore if event_ends_to_ignore else []
        super().__init__(
            event_starts_to_ignore=event_starts_to_ignore,
            event_ends_to_ignore=event_ends_to_ignore,
            logger=logger,
        )

    def on_event_start(
        self,
        event_type: CBEventType,
        payload: Optional[Dict[str, Any]] = None,
        event_id: str = "",
        parent_id: str = "",
        **kwargs: Any,
    ) -> str:
        """按事件类型存储事件开始数据。

Args:
    event_type (CBEventType): 要存储的事件类型。
    payload (Optional[Dict[str, Any]]): 要存储的有效负载。
    event_id (str): 要存储的事件ID。
    parent_id (str): 父事件ID。
"""
        event = CBEvent(event_type, payload=payload, id_=event_id)
        self._event_pairs_by_type[event.event_type].append(event)
        self._event_pairs_by_id[event.id_].append(event)
        self._sequential_events.append(event)
        return event.id_

    def on_event_end(
        self,
        event_type: CBEventType,
        payload: Optional[Dict[str, Any]] = None,
        event_id: str = "",
        **kwargs: Any,
    ) -> None:
        """按事件类型存储事件结束数据。

Args:
    event_type (CBEventType): 要存储的事件类型。
    payload (Optional[Dict[str, Any]]): 要存储的有效负载。
    event_id (str): 要存储的事件ID。
"""
        event = CBEvent(event_type, payload=payload, id_=event_id)
        self._event_pairs_by_type[event.event_type].append(event)
        self._event_pairs_by_id[event.id_].append(event)
        self._sequential_events.append(event)
        self._trace_map = defaultdict(list)

    def get_events(self, event_type: Optional[CBEventType] = None) -> List[CBEvent]:
        """获取特定事件类型的所有事件。"""
        if event_type is not None:
            return self._event_pairs_by_type[event_type]

        return self._sequential_events

    def _get_event_pairs(self, events: List[CBEvent]) -> List[List[CBEvent]]:
        """辅助函数，根据它们的ID将事件配对。"""
        event_pairs: Dict[str, List[CBEvent]] = defaultdict(list)
        for event in events:
            event_pairs[event.id_].append(event)

        return sorted(
            event_pairs.values(),
            key=lambda x: datetime.strptime(x[0].time, TIMESTAMP_FORMAT),
        )

    def _get_time_stats_from_event_pairs(
        self, event_pairs: List[List[CBEvent]]
    ) -> EventStats:
        """为一组事件对计算基于时间的统计信息。"""
        total_secs = 0.0
        for event_pair in event_pairs:
            start_time = datetime.strptime(event_pair[0].time, TIMESTAMP_FORMAT)
            end_time = datetime.strptime(event_pair[-1].time, TIMESTAMP_FORMAT)
            total_secs += (end_time - start_time).total_seconds()

        return EventStats(
            total_secs=total_secs,
            average_secs=total_secs / len(event_pairs),
            total_count=len(event_pairs),
        )

    def get_event_pairs(
        self, event_type: Optional[CBEventType] = None
    ) -> List[List[CBEvent]]:
        """通过ID将事件配对，可以是所有事件或特定类型的事件。"""
        if event_type is not None:
            return self._get_event_pairs(self._event_pairs_by_type[event_type])

        return self._get_event_pairs(self._sequential_events)

    def get_llm_inputs_outputs(self) -> List[List[CBEvent]]:
        """获取精确的LLM输入和输出。"""
        return self._get_event_pairs(self._event_pairs_by_type[CBEventType.LLM])

    def get_event_time_info(
        self, event_type: Optional[CBEventType] = None
    ) -> EventStats:
        event_pairs = self.get_event_pairs(event_type)
        return self._get_time_stats_from_event_pairs(event_pairs)

    def flush_event_logs(self) -> None:
        """清除内存中的所有事件。"""
        self._event_pairs_by_type = defaultdict(list)
        self._event_pairs_by_id = defaultdict(list)
        self._sequential_events = []

    def start_trace(self, trace_id: Optional[str] = None) -> None:
        """启动一个跟踪。"""
        self._trace_map = defaultdict(list)
        self._cur_trace_id = trace_id

    def end_trace(
        self,
        trace_id: Optional[str] = None,
        trace_map: Optional[Dict[str, List[str]]] = None,
    ) -> None:
        """关闭当前的跟踪。"""
        self._trace_map = trace_map or defaultdict(list)
        if self.print_trace_on_end:
            self.print_trace_map()

    def _print_trace_map(self, cur_event_id: str, level: int = 0) -> None:
        """递归打印跟踪地图到终端进行调试。"""
        event_pair = self._event_pairs_by_id[cur_event_id]
        if event_pair:
            time_stats = self._get_time_stats_from_event_pairs([event_pair])
            indent = " " * level * 2
            self._print(
                f"{indent}|_{event_pair[0].event_type} -> {time_stats.total_secs} seconds",
            )

        child_event_ids = self._trace_map[cur_event_id]
        for child_event_id in child_event_ids:
            self._print_trace_map(child_event_id, level=level + 1)

    def print_trace_map(self) -> None:
        """在终端打印简单的跟踪地图，用于调试最近的跟踪。"""
        self._print("*" * 10)
        self._print(f"Trace: {self._cur_trace_id}")
        self._print_trace_map(BASE_TRACE_EVENT, level=1)
        self._print("*" * 10)

    @property
    def event_pairs_by_type(self) -> Dict[CBEventType, List[CBEvent]]:
        return self._event_pairs_by_type

    @property
    def events_pairs_by_id(self) -> Dict[str, List[CBEvent]]:
        return self._event_pairs_by_id

    @property
    def sequential_events(self) -> List[CBEvent]:
        return self._sequential_events

on_event_start #

on_event_start(
    event_type: CBEventType,
    payload: Optional[Dict[str, Any]] = None,
    event_id: str = "",
    parent_id: str = "",
    **kwargs: Any
) -> str

按事件类型存储事件开始数据。

Parameters:

Name Type Description Default
event_type CBEventType

要存储的事件类型。

 required
payload Optional[Dict[str, Any]]

要存储的有效负载。

 None
event_id str

要存储的事件ID。

 ''
parent_id str

父事件ID。

 ''
Source code in llama_index/core/callbacks/llama_debug.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
    def on_event_start(
        self,
        event_type: CBEventType,
        payload: Optional[Dict[str, Any]] = None,
        event_id: str = "",
        parent_id: str = "",
        **kwargs: Any,
    ) -> str:
        """按事件类型存储事件开始数据。

Args:
    event_type (CBEventType): 要存储的事件类型。
    payload (Optional[Dict[str, Any]]): 要存储的有效负载。
    event_id (str): 要存储的事件ID。
    parent_id (str): 父事件ID。
"""
        event = CBEvent(event_type, payload=payload, id_=event_id)
        self._event_pairs_by_type[event.event_type].append(event)
        self._event_pairs_by_id[event.id_].append(event)
        self._sequential_events.append(event)
        return event.id_

on_event_end #

on_event_end(
    event_type: CBEventType,
    payload: Optional[Dict[str, Any]] = None,
    event_id: str = "",
    **kwargs: Any
) -> None

按事件类型存储事件结束数据。

Parameters:

Name Type Description Default
event_type CBEventType

要存储的事件类型。

 required
payload Optional[Dict[str, Any]]

要存储的有效负载。

 None
event_id str

要存储的事件ID。

 ''
Source code in llama_index/core/callbacks/llama_debug.py 
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
    def on_event_end(
        self,
        event_type: CBEventType,
        payload: Optional[Dict[str, Any]] = None,
        event_id: str = "",
        **kwargs: Any,
    ) -> None:
        """按事件类型存储事件结束数据。

Args:
    event_type (CBEventType): 要存储的事件类型。
    payload (Optional[Dict[str, Any]]): 要存储的有效负载。
    event_id (str): 要存储的事件ID。
"""
        event = CBEvent(event_type, payload=payload, id_=event_id)
        self._event_pairs_by_type[event.event_type].append(event)
        self._event_pairs_by_id[event.id_].append(event)
        self._sequential_events.append(event)
        self._trace_map = defaultdict(list)

get_events #

get_events(
    event_type: Optional[CBEventType] = None,
) -> List[CBEvent]

获取特定事件类型的所有事件。

Source code in llama_index/core/callbacks/llama_debug.py 
 95
 96
 97
 98
 99
100
def get_events(self, event_type: Optional[CBEventType] = None) -> List[CBEvent]:
    """获取特定事件类型的所有事件。"""
    if event_type is not None:
        return self._event_pairs_by_type[event_type]

    return self._sequential_events

get_event_pairs #

get_event_pairs(
    event_type: Optional[CBEventType] = None,
) -> List[List[CBEvent]]

通过ID将事件配对，可以是所有事件或特定类型的事件。

Source code in llama_index/core/callbacks/llama_debug.py 
129
130
131
132
133
134
135
136
def get_event_pairs(
    self, event_type: Optional[CBEventType] = None
) -> List[List[CBEvent]]:
    """通过ID将事件配对，可以是所有事件或特定类型的事件。"""
    if event_type is not None:
        return self._get_event_pairs(self._event_pairs_by_type[event_type])

    return self._get_event_pairs(self._sequential_events)

get_llm_inputs_outputs #

get_llm_inputs_outputs() -> List[List[CBEvent]]

获取精确的LLM输入和输出。

Source code in llama_index/core/callbacks/llama_debug.py 
138
139
140
def get_llm_inputs_outputs(self) -> List[List[CBEvent]]:
    """获取精确的LLM输入和输出。"""
    return self._get_event_pairs(self._event_pairs_by_type[CBEventType.LLM])

flush_event_logs #

flush_event_logs() -> None

清除内存中的所有事件。

Source code in llama_index/core/callbacks/llama_debug.py 
148
149
150
151
152
def flush_event_logs(self) -> None:
    """清除内存中的所有事件。"""
    self._event_pairs_by_type = defaultdict(list)
    self._event_pairs_by_id = defaultdict(list)
    self._sequential_events = []

start_trace #

start_trace(trace_id: Optional[str] = None) -> None

启动一个跟踪。

Source code in llama_index/core/callbacks/llama_debug.py 
154
155
156
157
def start_trace(self, trace_id: Optional[str] = None) -> None:
    """启动一个跟踪。"""
    self._trace_map = defaultdict(list)
    self._cur_trace_id = trace_id

end_trace #

end_trace(
    trace_id: Optional[str] = None,
    trace_map: Optional[Dict[str, List[str]]] = None,
) -> None

关闭当前的跟踪。

Source code in llama_index/core/callbacks/llama_debug.py 
159
160
161
162
163
164
165
166
167
def end_trace(
    self,
    trace_id: Optional[str] = None,
    trace_map: Optional[Dict[str, List[str]]] = None,
) -> None:
    """关闭当前的跟踪。"""
    self._trace_map = trace_map or defaultdict(list)
    if self.print_trace_on_end:
        self.print_trace_map()

print_trace_map #

print_trace_map() -> None

在终端打印简单的跟踪地图，用于调试最近的跟踪。

Source code in llama_index/core/callbacks/llama_debug.py 
183
184
185
186
187
188
def print_trace_map(self) -> None:
    """在终端打印简单的跟踪地图，用于调试最近的跟踪。"""
    self._print("*" * 10)
    self._print(f"Trace: {self._cur_trace_id}")
    self._print_trace_map(BASE_TRACE_EVENT, level=1)
    self._print("*" * 10)