清理日志数据

在将数据发送到日志集成（如langfuse等）之前，对消息进行编辑或掩盖个人身份信息（PII）。

设置自定义回调

from litellm.integrations.custom_logger import CustomLogger

class MyCustomHandler(CustomLogger):
    async def async_logging_hook(
        self, kwargs: dict, result: Any, call_type: str
    ) -> Tuple[dict, Any]:
        """
        用于掩盖日志请求/响应。返回请求/结果的修改版本。
        
        在`async_log_success_event`之前调用。
        """
        if (
            call_type == "completion" or call_type == "acompletion"
        ):  # /chat/completions 请求
            messages: Optional[List] = kwargs.get("messages", None)

            kwargs["messages"] = [{"role": "user", "content": "MASK_THIS_ASYNC_VALUE"}]

        return kwargs, responses

    def logging_hook(
        self, kwargs: dict, result: Any, call_type: str
    ) -> Tuple[dict, Any]:
        """
        用于掩盖日志请求/响应。返回请求/结果的修改版本。

        在`log_success_event`之前调用。
        """
        if (
            call_type == "completion" or call_type == "acompletion"
        ):  # /chat/completions 请求
            messages: Optional[List] = kwargs.get("messages", None)

            kwargs["messages"] = [{"role": "user", "content": "MASK_THIS_SYNC_VALUE"}]

        return kwargs, responses


customHandler = MyCustomHandler()

将自定义处理程序连接到LiteLLM

import litellm

litellm.callbacks = [customHandler]

测试它！

# pip install langfuse 

import os
import litellm
from litellm import completion 

os.environ["LANGFUSE_PUBLIC_KEY"] = ""
os.environ["LANGFUSE_SECRET_KEY"] = ""
# 可选，默认为 https://cloud.langfuse.com
os.environ["LANGFUSE_HOST"] # 可选
# LLM API密钥
os.environ['OPENAI_API_KEY']=""

litellm.callbacks = [customHandler]
litellm.success_callback = ["langfuse"]



## 同步
response = completion(model="gpt-3.5-turbo", messages=[{ "role": "user", "content": "Hi 👋 - i'm openai"}],
                              stream=True)
for chunk in response: 
    continue


## 异步
import asyncio 

def async completion():
    response = await acompletion(model="gpt-3.5-turbo", messages=[{ "role": "user", "content": "Hi 👋 - i'm openai"}],
                              stream=True)
    async for chunk in response: 
        continue
asyncio.run(completion())