第 25 期：监控与日志 (LLMOps)

LLMOps 全景

graph TB
    App[Dify App 运行] --> Monitor[监控层]
    Monitor --> Dashboard[📊 Dashboard]
    Monitor --> Logs[📋 日志追踪]
    Monitor --> Annotation[✅ 标注系统]
    
    Dashboard --> D1[请求量趋势]
    Dashboard --> D2[Token 消耗]
    Dashboard --> D3[响应延迟]
    Dashboard --> D4[用户活跃度]
    
    Logs --> L1[全链路追踪]
    Logs --> L2[节点级耗时]
    Logs --> L3[错误定位]
    
    Annotation --> A1[标注正确回答]
    Annotation --> A2[修正错误回答]
    Annotation --> A3[训练数据积累]

Dashboard 数据分析

# 通过 API 获取应用统计数据
def get_app_stats(app_id: str, days: int = 7):
    response = requests.get(
        f"{BASE_URL}/console/api/apps/{app_id}/statistics",
        headers={"Authorization": f"Bearer {CONSOLE_TOKEN}"},
        params={"days": days}
    )
    stats = response.json()
    
    print(f"=== 最近 {days} 天统计 ===")
    print(f"总请求数: {stats['total_messages']}")
    print(f"活跃用户: {stats['active_users']}")
    print(f"Token 消耗: {stats['total_tokens']:,}")
    print(f"平均延迟: {stats['avg_response_time']:.2f}s")
    print(f"错误率: {stats['error_rate']:.2%}")
    
    return stats

日志追踪

每条请求都有完整的执行日志，包含每个节点的输入、输出和耗时：

{
  "id": "log-xxx",
  "created_at": "2026-04-05T12:00:00Z",
  "query": "如何配置 SSL？",
  "answer": "您可以通过以下步骤...",
  "workflow_run": {
    "total_steps": 4,
    "total_tokens": 1250,
    "elapsed_time": 3.2,
    "steps": [
      {
        "node_id": "knowledge_retrieval_1",
        "node_type": "knowledge_retrieval",
        "status": "succeeded",
        "elapsed_time": 0.8,
        "outputs": {"segments_count": 3}
      },
      {
        "node_id": "llm_1",
        "node_type": "llm",
        "status": "succeeded",
        "elapsed_time": 2.1,
        "outputs": {"tokens": 850}
      }
    ]
  }
}

Annotation 标注系统

标注系统让你可以人工标记"好回答"和"坏回答"，用于持续优化。

sequenceDiagram
    participant User
    participant App as Dify App
    participant Annotator as 运营人员
    participant Cache as 标注缓存
    
    User->>App: "退货政策是什么？"
    App->>User: "我们支持7天无理由退货..."
    Annotator->>App: 标注此回答为 ✅ 优质回答
    App->>Cache: 缓存 (问题, 标准答案)
    
    User->>App: "怎么退货？" (类似问题)
    App->>Cache: 命中标注缓存
    Cache->>User: 直接返回标注的标准答案

# 通过 API 创建标注
def annotate_message(message_id: str, correct_answer: str):
    response = requests.post(
        f"{BASE_URL}/console/api/apps/{APP_ID}/annotations",
        headers={"Authorization": f"Bearer {CONSOLE_TOKEN}"},
        json={
            "message_id": message_id,
            "answer": correct_answer,
            "question": "退货政策是什么？"
        }
    )
    return response.json()