refactor(observability): Phase2.2 成本归因 trace_id列+token日志写入+LLM/MCP/媒体子span [v2026.07.19.3]

VERSION

@@ -3,11 +3,20 @@
 # 版本号格式: vYYYY.MM.DD.N   (年.月.日.当日递增序号，从 1 开始)
 # 环境标签:   prod = 正式版 | test = 测试版
 
-VERSION=v2026.07.19.1
+VERSION=v2026.07.19.3
 ENV=prod
 DATE=2026-07-19
 BASE_IMAGE=qihu-aiflow:with-deps
-BACKUP_IMAGE=qihu-aiflow:backup-20260719-001
+BACKUP_IMAGE=qihu-aiflow:backup-20260719-003
+
+## 本版本变更 (v2026.07.19.3 / prod)
+- Phase2.2 成本归因：token_usage_logs 新增 trace_id 列（迁移 020_add_trace_id_token_usage_logs.sql，幂等 ALTER），打通「链路 -> 成本」归集
+- _write_media_token_log（媒体图片/视频）注入 get_trace_id()，每条媒体 TokenUsageLog 现可回溯到请求链路
+- LogService.log_token_usage 增加 trace_id 参数（为后续文本 LLM token 记账接通成本归因预留）
+- chat_service / chat_media 在 LLM 补全、MCP 工具拉取/调用、技能执行、AGNES 媒体生成处加 span 子链路（async with span），一次请求可下钻耗时分布
+- 注意：_generate_media_task 为后台 fire-and-forget，其 span 会因处于 asyncio.create_task 表达式内而非法，已移除；真实媒体耗时由 chat_media 内 media.agnes_call span 覆盖
+- 纯增量埋点，不改接口返回结构与业务行为；pytest 9 passed + vitest 4 passed 门禁绿
+- 部署前已按 SOP 备份 rollback 点 qihu-aiflow:backup-20260719-003
 
 ## 本版本变更 (v2026.07.19.1 / prod)
 - Phase1 chat.py 巨石拆分收官：10070 行单体拆分为 chat_{utils,media,agents,skill,mcp,context,knowledge,service}.py 共 8 个子模块，controller 层 chat.py 降至 1352 行（<3000 目标达成）
@@ -16,6 +25,14 @@ BACKUP_IMAGE=qihu-aiflow:backup-20260719-001
 - 测试安全网守住 R1/R2/R3 不回退：后端 9 passed（smoke/market/sms + R1/R2/R3），前端 4 passed（datetime 时区格式化）；ci_gate 全绿
 - 部署前已按 SOP 备份 rollback 点 qihu-aiflow:backup-20260719-001（旧 chat.py 10070 行）
 
+## 本版本变更 (v2026.07.19.2 / prod)
+- Phase2.1 统一可观测基座：新增 app/core/observability.py（trace_id contextvar + TraceIdFilter + span 计时器 + 可选 JSON formatter），零侵入为全平台日志注入链路 ID
+- main.py APITrackingMiddleware 复用既有 request_id 作为 trace_id，并为每个响应注入 X-Trace-Id 响应头（前端可关联排查）
+- config.py setup_file_logging 接入 TraceIdFilter 并将日志格式升级为 [tid:%(trace_id)s]，所有 getLogger(__name__) 日志自动带链路 ID
+- chat.py send_message 路由用 span 包裹 _send_message_core（耗时+trace_id+conv_id/user_id），覆盖 LLM/MCP/媒体生成整体耗时，为后续成本归因建立回归基线
+- 纯增量埋点，不改任何接口返回结构与业务行为；向后兼容既有 logging 用法
+- 部署前已按 SOP 备份 rollback 点 qihu-aiflow:backup-20260719-002
+
 ## 历史
 - v2026.07.18.5 (prod)：Phase0 收官 — 最小 CI 门禁（ci_gate.py + pre-push 钩子 + 休眠 Gitea workflow）。后端 9 passed，前端 4 passed。
 - v2026.07.18.4 (prod)：前端 vitest 基建落地（4 passed，datetime 时区格式化）。

backend/python/app/api/chat.py

@@ -777,6 +777,8 @@ async def send_message(
 
 
 ):
+    from app.core.observability import span
+    async with span("chat.send_message", conv_id=conv_id, user_id=getattr(current_user, "id", None)):
         return await _send_message_core(conv_id, req, db, current_user)
 
 

backend/python/app/api/chat_media.py

@@ -47,6 +47,7 @@ from ..mcp_client import list_mcp_tools, call_mcp_tool
 from ..middleware.log_service import LogService
 from ..capability_registry import resolve_auto_model
 from app.api.chat_utils import (_html_escape, _proxy_for_url, cst_now, logger)
+from app.core.observability import span, get_trace_id
 
 def detect_media_intent(text: str) -> str | None:
 
@@ -958,6 +959,7 @@ async def _generate_media(db, ai_model, prompt: str, user_id: str = None) -> str
 
 
 
+        async with span("media.agnes_call"):
             resp, err = await _agnes_post_json(api_url, headers, payload, timeout)
 
 
@@ -1166,6 +1168,7 @@ async def _generate_media(db, ai_model, prompt: str, user_id: str = None) -> str
 
 
 
+        async with span("media.agnes_call"):
             resp, err = await _agnes_post_json(api_url, headers, payload, timeout)
 
 
@@ -1486,7 +1489,7 @@ async def _generate_media_task(ai_msg_id: str, ai_model_id: str, prompt: str, us
 
 
 
-        _write_media_token_log(db, ai_model, user_id, msg.conversation_id if msg else None, prompt)
+        _write_media_token_log(db, ai_model, user_id, msg.conversation_id if msg else None, prompt, trace_id=get_trace_id())
 
 
 
@@ -1516,7 +1519,7 @@ async def _generate_media_task(ai_msg_id: str, ai_model_id: str, prompt: str, us
 
             pass
 
-def _write_media_token_log(db, ai_model, user_id: str, conversation_id: str, prompt: str):
+def _write_media_token_log(db, ai_model, user_id: str, conversation_id: str, prompt: str, trace_id: str = None):
 
 
 
@@ -1601,6 +1604,7 @@ def _write_media_token_log(db, ai_model, user_id: str, conversation_id: str, pro
 
 
             created_at=cst_now(),
+            trace_id=trace_id,
 
 
 

backend/python/app/api/chat_service.py

@@ -16,6 +16,7 @@ import httpx
 import os
 import logging
 from ..database import get_db, SessionLocal
+from app.core.observability import span
 from ..models import (
 
 
@@ -791,6 +792,7 @@ async def _send_message_core(
 
                         try:
 
+                            async with span("mcp.list_tools", server=_srv.get("name") if isinstance(_srv, dict) else None):
                                 _tools = await list_mcp_tools(_srv)
 
                         except Exception as _e:
@@ -1192,6 +1194,7 @@ async def _send_message_core(
 
 
 
+                            async with span("llm.completion", model=ai_model.id if ai_model else None):
                                 resp = await client.post(
 
 
@@ -1380,6 +1383,7 @@ async def _send_message_core(
 
                                             if _srv:
 
+                                                async with span("mcp.call", name=_info.get("name") if isinstance(_info, dict) else None):
                                                     tool_result = await call_mcp_tool(
 
                                                         _srv, _info["name"], func_args
@@ -1390,6 +1394,7 @@ async def _send_message_core(
 
                                     else:
 
+                                        async with span("skill.execute", name=func_name):
                                             tool_result = _execute_skill_tool(
 
                                                 db, func_name, func_args, skills, req.content

backend/python/app/config.py

@@ -5,6 +5,7 @@ import os
 import logging
 from pathlib import Path
 from logging.handlers import TimedRotatingFileHandler
+from app.core.observability import TraceIdFilter, JsonFormatter
 from dotenv import load_dotenv
 
 # 加载 .env 文件（Docker 环境跳过，直接使用容器环境变量）
@@ -80,8 +81,12 @@ SLOW_REQUEST_THRESHOLD = float(os.getenv("SLOW_REQUEST_THRESHOLD", "3.0"))
 
 def setup_file_logging():
     """配置应用日志按天轮转输出到文件"""
+    use_json = os.getenv("AIFLOW_JSON_LOGS", "0") == "1"
+    if use_json:
+        log_format = JsonFormatter()
+    else:
         log_format = logging.Formatter(
-        "%(asctime)s [%(levelname)s] %(name)s - %(message)s",
+            "%(asctime)s [%(levelname)s] %(name)s [tid:%(trace_id)s] - %(message)s",
             datefmt="%Y-%m-%d %H:%M:%S",
         )
 
@@ -123,10 +128,12 @@ def setup_file_logging():
     root_logger.setLevel(logging.INFO)
     root_logger.addHandler(app_handler)
     root_logger.addHandler(error_handler)
+    root_logger.addFilter(TraceIdFilter())
 
     # 配置访问日志专用logger
     access_logger = logging.getLogger("aiflow.access")
     access_logger.addHandler(access_handler)
     access_logger.propagate = False
+    access_logger.addFilter(TraceIdFilter())
 
     return access_logger

backend/python/app/core/__init__.py

@@ -0,0 +1 @@
+# app/core package - 统一可观测基座 (Phase 2.1)

backend/python/app/core/observability.py

@@ -0,0 +1,174 @@
+"""
+Qihu-AIFlow 统一可观测基座 (Phase 2.1)
+=====================================
+提供贯穿一次请求的全局链路 ID（trace_id）与结构化日志能力，
+为后续重构（上下文引擎、媒体/LLM 成本归因）建立回归基线。
+
+设计原则（Strangler Fig 增量）：
+- 纯增量埋点，不改任何既有接口的返回结构。
+- 向后兼容现有 `logging.getLogger(__name__)` 用法：trace_id 由 root logger 的
+  TraceIdFilter 自动注入，调用方无需改动。
+- 不引入第三方依赖（structlog 未安装），自带可选的 JSON formatter。
+
+使用方式：
+    from app.core.observability import set_trace_id, get_trace_id, span
+
+    # 在中间件里为每个请求注入：
+    set_trace_id(request_id)
+
+    # 在关键调用处打计时 span（自动带 trace_id）：
+    async with span("chat.send_message", conv_id=conv_id):
+        return await _send_message_core(...)
+"""
+import logging
+import time
+import uuid
+import json
+from contextvars import ContextVar
+
+# ---------------------------------------------------------------------------
+# trace_id：贯穿一次请求（含其内所有子调用/协程）的全局链路 ID
+# ---------------------------------------------------------------------------
+_trace_id_var: ContextVar[str] = ContextVar("aiflow_trace_id", default="-")
+
+
+def new_trace_id() -> str:
+    """生成一个崭新的链路 ID（32 位 hex）。"""
+    return uuid.uuid4().hex
+
+
+def set_trace_id(tid: str | None = None) -> str:
+    """设置当前上下文的 trace_id；不传则自动生成。返回最终使用的 ID。"""
+    tid = tid or new_trace_id()
+    _trace_id_var.set(tid)
+    return tid
+
+
+def get_trace_id() -> str:
+    """读取当前上下文的 trace_id；未设置时返回占位 '-'。"""
+    return _trace_id_var.get()
+
+
+class TraceIdFilter(logging.Filter):
+    """为每一条日志记录注入 trace_id 属性（不存在则写为 '-'）。
+
+    挂载到 root logger 后，所有 `getLogger(__name__)` 的子 logger 都会继承，
+    从而让全平台既有日志自动带上链路 ID，零侵入。
+    """
+
+    def filter(self, record: logging.LogRecord) -> bool:
+        if not hasattr(record, "trace_id"):
+            try:
+                record.trace_id = get_trace_id()
+            except Exception:
+                record.trace_id = "-"
+        return True
+
+
+# 标准 logging 字段，JSON 序列化时需排除，避免与自定义 extra 冲突
+_STD_ATTRS = set(logging.makeLogRecord({}).__dict__.keys()) | {
+    "message", "asctime", "msg", "trace_id", "exc_info", "stack_info", "args",
+}
+
+
+class JsonFormatter(logging.Formatter):
+    """结构化 JSON 日志格式（通过环境变量 AIFLOW_JSON_LOGS=1 启用）。
+
+    默认不启用，避免破坏现有日志采集/人工查阅习惯。
+    """
+
+    def format(self, record: logging.LogRecord) -> str:
+        data = {
+            "ts": self.formatTime(record, "%Y-%m-%dT%H:%M:%S%z"),
+            "level": record.levelname,
+            "logger": record.name,
+            "trace_id": getattr(record, "trace_id", "-"),
+            "msg": record.getMessage(),
+        }
+        for k, v in record.__dict__.items():
+            if k not in _STD_ATTRS:
+                try:
+                    json.dumps(v, ensure_ascii=False)
+                    data[k] = v
+                except (TypeError, ValueError):
+                    data[k] = str(v)
+        if record.exc_info:
+            data["exc"] = self.formatException(record.exc_info)
+        return json.dumps(data, ensure_ascii=False)
+
+
+def get_logger(name: str | None = None) -> logging.Logger:
+    """获取 logger（向后兼容现有用法）。trace_id 由 root filter 自动注入。"""
+    return logging.getLogger(name)
+
+
+# ---------------------------------------------------------------------------
+# span：结构化计时上下文管理器（耗时 + trace_id + 自定义字段）
+# ---------------------------------------------------------------------------
+class _Span:
+    """同步/异步通用的计时 span。退出时记录 elapsed_ms 与状态。"""
+
+    def __init__(self, name: str, **fields):
+        self.name = name
+        self.fields = fields
+        self._logger = logging.getLogger("aiflow.span")
+        self._start = None
+        self.elapsed_ms = 0.0
+
+    def _enter(self):
+        self._start = time.perf_counter()
+        self._logger.info(
+            f"SPAN_START {self.name}",
+            extra={"span": self.name, "event": "start", "trace_id": get_trace_id(), **self.fields},
+        )
+        return self
+
+    def _exit(self, exc_type, exc, tb):
+        self.elapsed_ms = (time.perf_counter() - self._start) * 1000
+        status = "error" if exc_type else "ok"
+        self._logger.info(
+            f"SPAN_END {self.name} {self.elapsed_ms:.1f}ms",
+            extra={
+                "span": self.name,
+                "event": "end",
+                "elapsed_ms": round(self.elapsed_ms, 1),
+                "status": status,
+                "trace_id": get_trace_id(),
+                **self.fields,
+            },
+        )
+        return False  # 不吞掉异常
+
+    # 同步协议
+    def __enter__(self):
+        return self._enter()
+
+    def __exit__(self, exc_type, exc, tb):
+        return self._exit(exc_type, exc, tb)
+
+    # 异步协议
+    async def __aenter__(self):
+        return self._enter()
+
+    async def __aexit__(self, exc_type, exc, tb):
+        return self._exit(exc_type, exc, tb)
+
+
+def span(name: str, **fields):
+    """创建一个计时 span。可同步或异步 with。
+
+    async with span("chat.send_message", conv_id=cid):
+        ...
+    """
+    return _Span(name, **fields)
+
+
+__all__ = [
+    "new_trace_id",
+    "set_trace_id",
+    "get_trace_id",
+    "TraceIdFilter",
+    "JsonFormatter",
+    "get_logger",
+    "span",
+]

backend/python/app/middleware/log_service.py

@@ -57,7 +57,7 @@ class LogService:
                         app_id: str = None, app_name: str = None,
                         conversation_id: str = None, prompt_tokens: int = 0,
                         completion_tokens: int = 0, total_tokens: int = 0,
-                        cost: float = 0):
+                        cost: float = 0, trace_id: str = None):
         """记录Token消耗日志"""
         db = SessionLocal()
         try:
@@ -73,6 +73,7 @@ class LogService:
                 completion_tokens=completion_tokens,
                 total_tokens=total_tokens,
                 cost=cost,
+                trace_id=trace_id,
             )
             db.add(log)
             db.commit()

backend/python/app/models.py

@@ -537,6 +537,7 @@ class TokenUsageLog(Base):
     app_id = Column(String(36), default=None)
     app_name = Column(String(200), default=None)
     conversation_id = Column(String(36), default=None)
+    trace_id = Column(String(64), default=None)
     prompt_tokens = Column(Integer, default=0)
     completion_tokens = Column(Integer, default=0)
     total_tokens = Column(Integer, default=0)

backend/python/main.py

@@ -25,6 +25,7 @@ from app.api.admin.developer_tools_admin import router as developer_tools_router
 from app.api.admin.mcp_market_admin import router as mcp_market_router
 from app.api.admin import asset_admin, achievement_admin
 from app.config import setup_file_logging, SLOW_REQUEST_THRESHOLD
+from app.core.observability import set_trace_id
 from app.database import SessionLocal
 from app.models import OperationLog as OperationLogModel, TokenUsageLog as TokenUsageLogModel
 
@@ -41,6 +42,7 @@ class APITrackingMiddleware(BaseHTTPMiddleware):
 
     async def dispatch(self, request: Request, call_next):
         request_id = str(uuid.uuid4())
+        set_trace_id(request_id)
         start_time = time.time()
         path = request.url.path
         method = request.method
@@ -53,6 +55,7 @@ class APITrackingMiddleware(BaseHTTPMiddleware):
             response = await call_next(request)
             elapsed = time.time() - start_time
             status_code = response.status_code
+            response.headers["X-Trace-Id"] = request_id
 
             if not should_skip:
                 access_logger.info(

migrations/020_add_trace_id_token_usage_logs.sql

@@ -0,0 +1,7 @@
+-- Phase 2.2: add trace_id to token_usage_logs for cost attribution (idempotent)
+SET @db = DATABASE();
+SET @col_exists = (SELECT COUNT(*) FROM information_schema.columns WHERE table_schema = @db AND table_name = 'token_usage_logs' AND column_name = 'trace_id');
+SET @sql = IF(@col_exists = 0, 'ALTER TABLE token_usage_logs ADD COLUMN trace_id VARCHAR(64) NULL DEFAULT NULL', 'SELECT 1');
+PREPARE stmt FROM @sql;
+EXECUTE stmt;
+DEALLOCATE PREPARE stmt;