更新于

AI 流水线的错误追踪方案：从日志到根因定位

应用工程可观测性错误处理LLM 运维

AI 应用的错误追踪比传统 Web 服务更复杂：一条用户请求可能触发多次 LLM 调用、工具调用、向量检索，每一步都可能失败，而且失败方式千变万化——有时是 HTTP 错误，有时是模型”幻觉”出错误工具参数，有时是上下文长度溢出，有时是输出不符合预期格式。没有系统性的追踪方案，在生产环境里定位问题如同大海捞针。

本文从工程实践出发，梳理 AI 流水线错误追踪的核心设计，包括结构化日志、分布式 Trace、错误分类与告警，以及常见陷阱的避免方法。

一、AI 流水线错误的特殊性

与传统 API 服务相比，AI 流水线的错误有几个显著特点：

错误类型传统服务AI 流水线
错误来源代码逻辑、网络代码逻辑、网络 + 模型行为 + Prompt
确定性相同输入→相同错误相同输入→可能不同错误
可调试性复现容易随机性使复现困难
错误边界明确（异常/状态码）模糊（输出语义错误不抛异常）
延迟毫秒级秒级到分钟级

这些特点要求追踪系统不仅捕获”硬错误”（HTTP 500、超时），还要能感知”软错误”（输出格式不对、置信度过低、工具调用参数缺失）。

二、分层错误分类

建立清晰的错误分类是追踪的基础。推荐将 AI 流水线错误分为四层：

from enum import Enum

class ErrorLayer(Enum):
    INFRASTRUCTURE = "infra"      # 网络、超时、认证
    API = "api"                   # LLM 服务端错误（429、500）
    PROTOCOL = "protocol"         # 响应格式不符合预期（JSON 解析失败）
    SEMANTIC = "semantic"         # 语义错误（工具参数错误、回答与问题不符）

class PipelineError(Exception):
    def __init__(
        self,
        layer: ErrorLayer,
        step: str,
        message: str,
        retryable: bool = False,
        context: dict | None = None,
    ):
        self.layer = layer
        self.step = step
        self.retryable = retryable
        self.context = context or {}
        super().__init__(f"[{layer.value}:{step}] {message}")

这种分类让下游处理器能快速判断：是否需要重试（infra/api 层通常可重试）、是否需要告警（semantic 层错误往往说明 Prompt 设计有问题）、是否需要人工干预（protocol 层频繁出错说明输出格式控制需要加强）。

三、结构化日志设计

AI 流水线的日志需要比传统服务携带更多上下文。推荐的日志结构：

import json
import time
import uuid
import logging
from dataclasses import dataclass, asdict
from typing import Any

@dataclass
class PipelineLogEntry:
    trace_id: str           # 贯穿整条请求的唯一 ID
    span_id: str            # 当前步骤的唯一 ID
    parent_span_id: str | None  # 父步骤 ID（用于构建调用树）
    step_name: str          # 步骤名称，如 "llm_call", "tool_search"
    model: str | None       # 使用的模型
    start_time: float
    end_time: float
    latency_ms: float
    status: str             # "ok" / "error" / "retry"
    error_layer: str | None
    error_message: str | None
    input_tokens: int | None
    output_tokens: int | None
    prompt_hash: str | None  # Prompt 的哈希，便于聚合相同 Prompt 的错误
    metadata: dict[str, Any]

class StructuredLogger:
    def __init__(self, service_name: str):
        self.service_name = service_name
        self.logger = logging.getLogger(service_name)
    
    def log_span(self, entry: PipelineLogEntry):
        record = {
            "service": self.service_name,
            **asdict(entry),
        }
        # 输出 JSON 日志，便于 ELK/Loki 等日志系统解析
        self.logger.info(json.dumps(record, ensure_ascii=False))

关键字段说明：

  • trace_id：用户一次请求从进入到结束，全程共享同一个 trace_id，跨多个 LLM 调用
  • prompt_hash：对 Prompt 内容做 MD5，便于在告警系统中聚合”哪个 Prompt 模板出错率高”
  • input_tokens / output_tokens：Token 消耗追踪，既用于成本分析，也用于发现上下文溢出

四、Span 追踪：构建调用树

AI 流水线通常是树状结构（一步 LLM 调用可能触发多个工具调用，每个工具调用又可能触发子 LLM 调用）。用 Span 树来表示这个结构：

import contextlib
import hashlib

class PipelineTracer:
    def __init__(self, logger: StructuredLogger):
        self.logger = logger
        self._current_trace_id: str | None = None
        self._span_stack: list[str] = []
    
    def new_trace(self) -> str:
        self._current_trace_id = str(uuid.uuid4())
        self._span_stack = []
        return self._current_trace_id
    
    @contextlib.contextmanager
    def span(
        self,
        step_name: str,
        model: str | None = None,
        metadata: dict | None = None,
    ):
        span_id = str(uuid.uuid4())[:8]
        parent_span_id = self._span_stack[-1] if self._span_stack else None
        self._span_stack.append(span_id)
        
        start_time = time.time()
        entry = PipelineLogEntry(
            trace_id=self._current_trace_id or "no-trace",
            span_id=span_id,
            parent_span_id=parent_span_id,
            step_name=step_name,
            model=model,
            start_time=start_time,
            end_time=0,
            latency_ms=0,
            status="ok",
            error_layer=None,
            error_message=None,
            input_tokens=None,
            output_tokens=None,
            prompt_hash=None,
            metadata=metadata or {},
        )
        
        try:
            yield entry
            entry.status = "ok"
        except PipelineError as e:
            entry.status = "error"
            entry.error_layer = e.layer.value
            entry.error_message = str(e)
            raise
        except Exception as e:
            entry.status = "error"
            entry.error_layer = ErrorLayer.INFRASTRUCTURE.value
            entry.error_message = str(e)
            raise
        finally:
            entry.end_time = time.time()
            entry.latency_ms = (entry.end_time - start_time) * 1000
            self._span_stack.pop()
            self.logger.log_span(entry)

使用示例：

tracer = PipelineTracer(StructuredLogger("my-ai-app"))

async def run_pipeline(user_query: str) -> str:
    trace_id = tracer.new_trace()
    
    with tracer.span("understand_intent", model="gpt-4o-mini") as span:
        # 第一步：意图理解
        intent = await call_llm_for_intent(user_query)
        span.metadata["detected_intent"] = intent
    
    with tracer.span("retrieve_context") as span:
        # 第二步：向量检索
        docs = await vector_search(intent)
        span.metadata["retrieved_docs"] = len(docs)
    
    with tracer.span("generate_answer", model="gpt-4o") as span:
        # 第三步：生成回答
        answer = await call_llm_for_answer(user_query, docs)
        span.metadata["answer_length"] = len(answer)
    
    return answer

五、LLM 调用的专用追踪包装器

LLM 调用是 AI 流水线最核心的步骤，值得单独封装：

from openai import OpenAI, APIError, RateLimitError, APITimeoutError
import hashlib

def hash_prompt(messages: list) -> str:
    content = json.dumps(messages, ensure_ascii=False, sort_keys=True)
    return hashlib.md5(content.encode()).hexdigest()[:8]

def traced_llm_call(
    client: OpenAI,
    tracer: PipelineTracer,
    step_name: str,
    model: str,
    messages: list,
    max_retries: int = 3,
    **kwargs,
) -> str:
    prompt_hash = hash_prompt(messages)
    
    for attempt in range(max_retries):
        with tracer.span(step_name, model=model, metadata={"attempt": attempt, "prompt_hash": prompt_hash}) as span:
            try:
                response = client.chat.completions.create(
                    model=model,
                    messages=messages,
                    **kwargs,
                )
                usage = response.usage
                span.input_tokens = usage.prompt_tokens if usage else None
                span.output_tokens = usage.completion_tokens if usage else None
                span.prompt_hash = prompt_hash
                return response.choices[0].message.content
            
            except RateLimitError as e:
                if attempt < max_retries - 1:
                    wait = 2 ** attempt
                    span.status = "retry"
                    time.sleep(wait)
                    continue
                raise PipelineError(
                    ErrorLayer.API, step_name,
                    f"Rate limit after {max_retries} retries",
                    retryable=False,
                )
            
            except APITimeoutError as e:
                raise PipelineError(
                    ErrorLayer.INFRASTRUCTURE, step_name,
                    "LLM call timed out",
                    retryable=True,
                )
            
            except APIError as e:
                raise PipelineError(
                    ErrorLayer.API, step_name,
                    f"API error {e.status_code}: {e.message}",
                    retryable=e.status_code >= 500,
                )

六、软错误检测：捕获语义层失败

硬错误（异常）容易捕获，软错误（输出不符合预期）是 AI 应用的特殊挑战。

import json as json_module
from pydantic import BaseModel, ValidationError

class ToolCallOutput(BaseModel):
    tool_name: str
    arguments: dict
    reasoning: str | None = None

def parse_and_validate_output(
    raw_output: str,
    step_name: str,
    tracer: PipelineTracer,
) -> ToolCallOutput:
    with tracer.span(f"{step_name}.parse_output") as span:
        # 尝试解析 JSON
        try:
            data = json_module.loads(raw_output)
        except json_module.JSONDecodeError as e:
            span.status = "error"
            span.error_layer = ErrorLayer.PROTOCOL.value
            span.metadata["raw_output_preview"] = raw_output[:200]
            raise PipelineError(
                ErrorLayer.PROTOCOL, step_name,
                f"Output is not valid JSON: {e}",
                context={"raw_output": raw_output[:500]},
            )
        
        # 验证结构
        try:
            result = ToolCallOutput(**data)
            span.metadata["tool_name"] = result.tool_name
            return result
        except ValidationError as e:
            span.status = "error"
            span.error_layer = ErrorLayer.SEMANTIC.value
            span.metadata["validation_errors"] = str(e)
            raise PipelineError(
                ErrorLayer.SEMANTIC, step_name,
                f"Output structure invalid: {e}",
                context={"parsed_data": data},
            )

七、错误聚合与告警设计

单条日志可以定位单次问题，聚合分析才能发现系统性问题。推荐关注以下指标：

# 伪代码：告警规则示例（实际接入 Prometheus / Datadog / Grafana）

ALERT_RULES = {
    # 规则 1：特定步骤错误率超过阈值
    "step_error_rate": {
        "query": "rate(pipeline_errors_total[5m]) / rate(pipeline_requests_total[5m])",
        "threshold": 0.05,  # 5%
        "severity": "warning",
    },
    # 规则 2：同一 prompt_hash 连续错误（Prompt 设计问题）
    "prompt_hash_consecutive_errors": {
        "query": "consecutive_errors_by_prompt_hash > 5",
        "severity": "critical",
        "action": "pause_and_review_prompt",
    },
    # 规则 3：平均延迟突增（模型服务降级信号）
    "latency_spike": {
        "query": "avg_latency_ms > baseline_latency_ms * 3",
        "severity": "warning",
    },
    # 规则 4：Token 消耗异常（防止意外的上下文膨胀）
    "token_spike": {
        "query": "avg_input_tokens > expected_tokens * 5",
        "severity": "warning",
        "action": "check_context_accumulation",
    },
}

关键告警场景

  1. 语义层错误率高（ErrorLayer.SEMANTIC 超过 5%）：通常意味着 Prompt 模板需要修改，或输出格式约束需要加强
  2. 特定 prompt_hash 反复出错：说明某个具体的 Prompt 有设计问题，需要人工 review
  3. input_tokens 异常增长：可能是 Agent 在循环追加上下文，需要检查 Agent 循环逻辑
  4. 重试率突增：往往是 LLM 服务商的限速或稳定性问题，考虑切换备用模型

八、生产环境最佳实践清单

  • 永远记录 trace_id：每次请求生成唯一 ID，贯穿所有子调用，用户反馈问题时能直接 grep 出全链路日志
  • 记录 Prompt 哈希而非 Prompt 全文：Prompt 可能包含用户隐私数据，只存哈希用于聚合分析
  • 区分可重试与不可重试错误：速率限制（429）可重试，认证错误（401）不可重试，盲目重试会放大问题
  • 软错误要设阈值：不是每次软错误都要告警，连续 N 次或错误率超过 X% 才触发
  • 保留最近 N 次错误的完整上下文：内存中 ring buffer，用于 debug，不持久化到日志（避免隐私问题）
  • 使用 API 中转服务时关注其错误码映射：不同中转服务的错误码与原始 API 可能有差异，建议查阅文档

AI 应用的错误处理设计可进一步参考 AI Agent 错误恢复策略，那篇文章重点讨论了多步 Agent 场景下的回滚与重试机制。

九、相关阅读

构建生产级 AI 流水线时，YoTradeApi 提供统一 API 中转，支持透传 OpenAI 标准错误码，方便与上述追踪方案集成，实现多模型的统一可观测性。

相关阅读