langfuse追踪Trace

介绍

🧠 Langfuse 是什么？

Langfuse 是一个专门为 LLM 应用（如 OpenAI / LangChain / 自定义 Agent） 设计的 观测与追踪平台（Observability Platform）。

简单说，它就像是你为 AI 应用插上的 “黑匣子”，可以记录每一次调用过程、上下文、耗时、错误等信息，并提供 Web UI 让你可视化分析和优化。

相关界面：

使用

注册langfuse类

本来想使用装饰器的，发现 没什么用，还是用langfuse类 去实现吧

from langfuse import Langfuse
import os
from functools import wraps

class LangfuseService:
    def __init__(self):
        self.langfuse = Langfuse(
            public_key=os.getenv("LANGFUSE_PUBLIC_KEY"),
            secret_key=os.getenv("LANGFUSE_SECRET_KEY"),
            host=os.getenv("LANGFUSE_HOST")
        )

    def trace_langfuse_generator(self, name="default-trace", user_id="anonymous"):
        def decorator(func):
            @wraps(func)
            async def wrapper(*args, **kwargs):
                trace = self.langfuse.trace(name=name, user_id=user_id)
                span = trace.span(name=func.__name__)
                try:
                    async for chunk in func(*args, **kwargs):
                        yield chunk
                    span.end()
                except Exception as e:
                    span.end(output=f"Error: {str(e)}")
                    raise
            return wrapper
        return decorator
        
    
langfuse_service = LangfuseService()

在gpt类里面：

注意默认不支持流式的捕捉，所以需要在+= 捕捉流式的总输出

async def gpt_stream_with_langfuse(self, user_message: str, model: str = None, history: list[dict] = [], system_prompt: str = ""):
        trace = langfuse_service.langfuse.trace(name="gpt_stream", user_id="user-123")
        span = trace.span(name="gpt_stream_call")

        try:
            messages = []
            if history:
                messages.extend(history)

            if system_prompt:
                messages.append({"role": "system", "content": system_prompt})

            messages.append({"role": "user", "content": user_message})

            # 🔍 添加 input
            span.update(input={"messages": messages})

            response = await self.oai_client.chat.completions.create(
                model=model or self.model,
                messages=messages,
                stream=True
            )

            full_response = ""
            async for chunk in response:
                if chunk.choices[0].delta.content:
                    full_response += chunk.choices[0].delta.content
                    yield chunk.choices[0].delta.content

            # 🔍 添加输出
            span.end(output=full_response)
            trace.update(output=full_response)
        except Exception as e:
            span.end(output={"error": str(e)})
            trace.update(output={"error": str(e)})
            raise

创建一个 Trace 对象，用于记录一次完整的 LLM 调用。

Trace

Trace 是一个“顶层调用上下文”，可以理解为：

• 用户发起了一次聊天、搜索、问答请求

• 系统执行了一系列动作（调用 OpenAI、函数工具、数据库等）

• 我们希望把这整个过程的输入输出、耗时、状态都记录下来

✅ 返回值：

返回的是一个 StatefulTraceClient 对象，后续你可以通过它：

• 添加 子步骤 span

• 更新 trace 内容（例如设置最终输出）

• 获取 Langchain 回调器（如果用 Langchain）

返回的是一个 StatefulTraceClient 对象，后续你可以通过它：

• 添加 子步骤 span

• 更新 trace 内容（例如设置最终输出）

• 获取 Langchain 回调器（如果用 Langchain）

Span

创建一个 Span 子操作，用于追踪 Trace 内部的一个子任务/子步骤。

📘 Span 是什么？

• 是 Trace 中的一个步骤

• 有自己的 start / end 生命周期

• 可以添加 input / output / 错误等信息

trace = langfuse_service.langfuse.trace(name="gpt_stream", user_id="user-123")
span = trace.span(name="gpt_stream_call", input={"question": "北京的时间是多少？"})

try:
    # 你的业务逻辑
    result = await openai_call()  # 比如调用 openai 的 chat 接口
    span.end(output=result)       # 结束 span，并记录结果
    trace.update(output=result)   # 更新整个 trace 的 output
except Exception as e:
    span.end(output=f"Error: {str(e)}")
    raise