Openinference

OpenInferenceCallbackHandler

Bases: BaseCallbackHandler

用于将生成数据存储为OpenInference格式的回调处理程序。 OpenInference是用于捕获和存储AI模型推断的开放标准。它使生产LLMapp服务器能够与LLM可观测性解决方案（如Arize和Phoenix）无缝集成。

有关规范的更多信息，请参见 https://github.com/Arize-ai/open-inference-spec

Source code in llama_index/callbacks/openinference/base.py

class OpenInferenceCallbackHandler(BaseCallbackHandler):
    """用于将生成数据存储为OpenInference格式的回调处理程序。
    OpenInference是用于捕获和存储AI模型推断的开放标准。它使生产LLMapp服务器能够与LLM可观测性解决方案（如Arize和Phoenix）无缝集成。

    有关规范的更多信息，请参见
    https://github.com/Arize-ai/open-inference-spec"""

    def __init__(
        self,
        callback: Optional[Callable[[List[QueryData], List[NodeData]], None]] = None,
    ) -> None:
        """初始化OpenInferenceCallbackHandler。

Args:
    callback（可选[Callable[[List[QueryData], List[NodeData]], None]，可选）：一个回调函数，当查询跟踪完成时将被调用，通常用于记录或持久化查询数据。
"""
        super().__init__(event_starts_to_ignore=[], event_ends_to_ignore=[])
        self._callback = callback
        self._trace_data = TraceData()
        self._query_data_buffer: List[QueryData] = []
        self._node_data_buffer: List[NodeData] = []

    def start_trace(self, trace_id: Optional[str] = None) -> None:
        if trace_id == "query" or trace_id == "chat":
            self._trace_data = TraceData()
            self._trace_data.query_data.timestamp = datetime.now().isoformat()
            self._trace_data.query_data.id = _generate_random_id()

    def end_trace(
        self,
        trace_id: Optional[str] = None,
        trace_map: Optional[Dict[str, List[str]]] = None,
    ) -> None:
        if trace_id == "query" or trace_id == "chat":
            self._query_data_buffer.append(self._trace_data.query_data)
            self._node_data_buffer.extend(self._trace_data.node_datas)
            self._trace_data = TraceData()
            if self._callback is not None:
                self._callback(self._query_data_buffer, self._node_data_buffer)

    def on_event_start(
        self,
        event_type: CBEventType,
        payload: Optional[Dict[str, Any]] = None,
        event_id: str = "",
        parent_id: str = "",
        **kwargs: Any,
    ) -> str:
        if payload is not None:
            if event_type is CBEventType.QUERY:
                query_text = payload[EventPayload.QUERY_STR]
                self._trace_data.query_data.query_text = query_text
            elif event_type is CBEventType.LLM:
                if prompt := payload.get(EventPayload.PROMPT, None):
                    self._trace_data.query_data.llm_prompt = prompt
                if messages := payload.get(EventPayload.MESSAGES, None):
                    self._trace_data.query_data.llm_messages = [
                        (m.role.value, m.content) for m in messages
                    ]
                    # For chat engines there is no query event and thus the
                    # query text will be None, in this case we set the query
                    # text to the last message passed to the LLM
                    if self._trace_data.query_data.query_text is None:
                        self._trace_data.query_data.query_text = messages[-1].content
        return event_id

    def on_event_end(
        self,
        event_type: CBEventType,
        payload: Optional[Dict[str, Any]] = None,
        event_id: str = "",
        **kwargs: Any,
    ) -> None:
        if payload is None:
            return
        if event_type is CBEventType.RETRIEVE:
            for node_with_score in payload[EventPayload.NODES]:
                node = node_with_score.node
                score = node_with_score.score
                self._trace_data.query_data.node_ids.append(node.hash)
                self._trace_data.query_data.scores.append(score)
                self._trace_data.node_datas.append(
                    NodeData(
                        id=node.hash,
                        node_text=node.text,
                    )
                )
        elif event_type is CBEventType.LLM:
            if self._trace_data.query_data.response_text is None:
                if response := payload.get(EventPayload.RESPONSE, None):
                    if isinstance(response, ChatResponse):
                        # If the response is of class ChatResponse the string
                        # representation has the format "<role>: <message>",
                        # but we want just the message
                        response_text = response.message.content
                    else:
                        response_text = str(response)
                    self._trace_data.query_data.response_text = response_text
                elif completion := payload.get(EventPayload.COMPLETION, None):
                    self._trace_data.query_data.response_text = str(completion)
        elif event_type is CBEventType.EMBEDDING:
            self._trace_data.query_data.query_embedding = payload[
                EventPayload.EMBEDDINGS
            ][0]

    def flush_query_data_buffer(self) -> List[QueryData]:
        """清除查询数据缓冲区并返回数据。

返回：
    List[QueryData]: 查询数据。
"""
        query_data_buffer = self._query_data_buffer
        self._query_data_buffer = []
        return query_data_buffer

    def flush_node_data_buffer(self) -> List[NodeData]:
        """清除节点数据缓冲区并返回数据。

返回：
    List[NodeData]: 节点数据。
"""
        node_data_buffer = self._node_data_buffer
        self._node_data_buffer = []
        return node_data_buffer

flush_query_data_buffer

flush_query_data_buffer() -> List[QueryData]

清除查询数据缓冲区并返回数据。

返回： List[QueryData]: 查询数据。

Source code in llama_index/callbacks/openinference/base.py

    def flush_query_data_buffer(self) -> List[QueryData]:
        """清除查询数据缓冲区并返回数据。

返回：
    List[QueryData]: 查询数据。
"""
        query_data_buffer = self._query_data_buffer
        self._query_data_buffer = []
        return query_data_buffer

flush_node_data_buffer

flush_node_data_buffer() -> List[NodeData]

清除节点数据缓冲区并返回数据。

返回： List[NodeData]: 节点数据。

Source code in llama_index/callbacks/openinference/base.py

    def flush_node_data_buffer(self) -> List[NodeData]:
        """清除节点数据缓冲区并返回数据。

返回：
    List[NodeData]: 节点数据。
"""
        node_data_buffer = self._node_data_buffer
        self._node_data_buffer = []
        return node_data_buffer