Langfuse三层追踪结构详解与应用

import timeimport uuidfrom typing import List, Dict, Optionalfrom dataclasses import dataclass, fieldfrom datetime import datetime  @dataclassclass Observation:    """最细粒度追踪单元：一次LLM API调用或工具调用"""    id: str = field(default_factory=lambda: str(uuid.uuid4())[:8])    name: str = ""    parent_span_id: Optional[str] = None    model: str = ""    input_prompt: str = ""    output_response: str = ""    prompt_tokens: int = 0    completion_tokens: int = 0    total_tokens: int = 0    cost: float = 0.0    latency_ms: float = 0.0    start_time: float = 0.0    end_time: float = 0.0    metadata: Dict = field(default_factory=dict)     def to_dict(self) -> Dict:        return {            "id": self.id, "name": self.name, "type": "observation",            "parent_span_id": self.parent_span_id,            "model": self.model, "input": self.input_prompt[:50] + "...",            "output": self.output_response[:50] + "...",            "usage": {"prompt_tokens": self.prompt_tokens,                      "completion_tokens": self.completion_tokens,                      "total_tokens": self.total_tokens},            "cost": self.cost, "latency_ms": round(self.latency_ms, 1)        }  @dataclassclass Span:    """中间层追踪单元：一个逻辑处理步骤"""    id: str = field(default_factory=lambda: str(uuid.uuid4())[:8])    name: str = ""    trace_id: str = ""    parent_span_id: Optional[str] = None    start_time: float = 0.0    end_time: float = 0.0    observations: List[Observation] = field(default_factory=list)    child_spans: List["Span"] = field(default_factory=list)    metadata: Dict = field(default_factory=dict)     @property    def duration_ms(self) -> float:        return (self.end_time - self.start_time) * 1000     def add_observation(self, obs: Observation):        obs.parent_span_id = self.id        self.observations.append(obs)     def add_child_span(self, child: "Span"):        child.parent_span_id = self.id        self.child_spans.append(child)     def to_dict(self) -> Dict:        return {            "id": self.id, "name": self.name, "type": "span",            "parent_span_id": self.parent_span_id,            "duration_ms": round(self.duration_ms, 1),            "observations": [o.to_dict() for o in self.observations],            "child_spans": [s.to_dict() for s in self.child_spans]        }  @dataclassclass Trace:    """顶层追踪单元：一次用户请求的完整处理生命周期"""    id: str = field(default_factory=lambda: str(uuid.uuid4())[:8])    name: str = ""    session_id: str = ""    user_id: str = ""    start_time: float = 0.0    end_time: float = 0.0    status: str = "success"    spans: List[Span] = field(default_factory=list)    metadata: Dict = field(default_factory=dict)    total_cost: float = 0.0     @property    def duration_ms(self) -> float:        return (self.end_time - self.start_time) * 1000     def add_span(self, span: Span):        span.trace_id = self.id        self.spans.append(span)     def compute_total_cost(self) -> float:        total = 0.0        for span in self.spans:            for obs in span.observations:                total += obs.cost            for child in span.child_spans:                for obs in child.observations:                    total += obs.cost        self.total_cost = total        return total     def to_dict(self) -> Dict:        return {            "id": self.id, "name": self.name, "type": "trace",            "session_id": self.session_id, "user_id": self.user_id,            "duration_ms": round(self.duration_ms, 1),            "status": self.status, "total_cost": round(self.total_cost, 4),            "spans": [s.to_dict() for s in self.spans]        }  class LangfuseTracer:    """Langfuse三级追踪器：管理trace/span/observation的创建和关联"""     def __init__(self):        self.traces: List[Trace] = []     def create_trace(self, name: str, session_id: str = "",                     user_id: str = "") -> Trace:        trace = Trace(name=name, session_id=session_id, user_id=user_id,                      start_time=time.time())        self.traces.append(trace)        return trace     def create_span(self, trace: Trace, name: str,                    parent: Optional[Span] = None) -> Span:        span = Span(name=name, trace_id=trace.id, start_time=time.time())        if parent:            parent.add_child_span(span)        else:            trace.add_span(span)        return span     def create_observation(self, span: Span, name: str, model: str,                           input_prompt: str, output: str,                           prompt_tokens: int, completion_tokens: int,                           cost_per_1k: float = 0.002) -> Observation:        start = time.time()        total_tokens = prompt_tokens + completion_tokens        cost = (total_tokens / 1000.0) * cost_per_1k        obs = Observation(            name=name, model=model, input_prompt=input_prompt,            output_response=output, prompt_tokens=prompt_tokens,            completion_tokens=completion_tokens, total_tokens=total_tokens,            cost=cost, start_time=start,            end_time=start + 0.05 + total_tokens * 0.0001        )        span.add_observation(obs)        return obs     def finish_span(self, span: Span):        span.end_time = time.time()     def finish_trace(self, trace: Trace):        trace.end_time = time.time()        trace.compute_total_cost()     def analyze_bottleneck(self, trace: Trace) -> Dict:        """性能瓶颈分析：找出耗时最长的span"""        span_durations = []        for span in trace.spans:            span_durations.append({                "name": span.name, "duration_ms": round(span.duration_ms, 1),                "pct": 0.0            })            for child in span.child_spans:                span_durations.append({                    "name": f"  └─{child.name}",                    "duration_ms": round(child.duration_ms, 1),                    "pct": 0.0                })        total = max(trace.duration_ms, 0.001)        for s in span_durations:            s["pct"] = round(s["duration_ms"] / total * 100, 1)        span_durations.sort(key=lambda x: x["duration_ms"], reverse=True)        return {"total_ms": round(total, 1), "bottleneck": span_durations[0],                "breakdown": span_durations}  def demo_langfuse_tracing():    """演示Langfuse三级追踪结构在RAG应用中的使用"""    tracer = LangfuseTracer()     print("=" * 60)    print("Langfuse 三级追踪结构演示：RAG应用调用链")    print("=" * 60)     trace = tracer.create_trace("RAG Query", session_id="sess_001",                                user_id="user_42")    print(f"\n[Trace] {trace.name} (session={trace.session_id})")     # Span 1: 检索步骤    retrieve_span = tracer.create_span(trace, "retrieve_documents")    print(f"  [Span] {retrieve_span.name}")     obs1 = tracer.create_observation(        retrieve_span, "vector_search", "text-embedding-3-small",        "What is machine learning?", "[0.12, -0.34, ...]",        prompt_tokens=8, completion_tokens=0, cost_per_1k=0.0001    )    print(f"    [Observation] {obs1.name}: model={obs1.model}, "          f"tokens={obs1.total_tokens}, cost=${obs1.cost:.6f}")     obs2 = tracer.create_observation(        retrieve_span, "rerank", "bge-reranker-v2",        "5 candidate documents", "Top 3 documents selected",        prompt_tokens=500, completion_tokens=0, cost_per_1k=0.001    )    print(f"    [Observation] {obs2.name}: model={obs2.model}, "          f"tokens={obs2.total_tokens}, cost=${obs2.cost:.6f}")    tracer.finish_span(retrieve_span)     # Span 2: 生成步骤    generate_span = tracer.create_span(trace, "generate_answer")    print(f"  [Span] {generate_span.name}")     obs3 = tracer.create_observation(        generate_span, "llm_generation", "gpt-4o",        "Context: ... Question: What is ML?",        "Machine learning is a subset of AI...",        prompt_tokens=1200, completion_tokens=350, cost_per_1k=0.005    )    print(f"    [Observation] {obs3.name}: model={obs3.model}, "          f"tokens={obs3.total_tokens}, cost=${obs3.cost:.6f}")    tracer.finish_span(generate_span)     tracer.finish_trace(trace)     # 性能分析    print(f"\n{'=' * 60}")    print("性能瓶颈分析")    print(f"{'=' * 60}")    analysis = tracer.analyze_bottleneck(trace)    print(f"总耗时: {analysis['total_ms']}ms")    for item in analysis["breakdown"]:        print(f"  {item['name']}: {item['duration_ms']}ms ({item['pct']}%)")     print(f"\n总成本: ${trace.total_cost:.6f}")    print(f"Trace状态: {trace.status}")  if __name__ == "__main__":    demo_langfuse_tracing()