Compare commits

...
20 changed files with 788 additions and 76 deletions

No files matched your search

+1 -26
View File
@@ -1,26 +1 @@
# 七虎AI PaaS 平台 — 版本清单 (VERSION)
# 版本规范见 docs/VERSIONING.md
# 版本号格式: vYYYY.MM.DD.N (年.月.日.当日递增序号,从 1 开始)
# 环境标签: prod = 正式版 | test = 测试版
VERSION=v2026.07.19.1
ENV=prod
DATE=2026-07-19
BASE_IMAGE=qihu-aiflow:with-deps
BACKUP_IMAGE=qihu-aiflow:backup-20260719-001
## 本版本变更 (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 目标达成)
- 路由处理器(/conversations、/messages、/media/relay、/models 等全部 10 个路由)保留在 chat.py,业务/服务函数迁至子模块并通过 import 再导出
- 关键修复:此前拆分误删了全部 @router 装饰器导致路由为 0(重启即全 404),本版本以 AST 比对原始提交恢复装饰器,并补回 chat_service._send_message_core 导入
- 测试安全网守住 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.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 时区格式化)。
- v2026.07.18.3 (prod):回归用例 R1/R2/R3(9 passed)— 微信Bot截断、技能/MCP覆盖规则、媒体生成超时回写。
- v2026.07.18.2 (prod):后端 pytest 基建落地(6 passed)— 冒烟/市场14类/短信宝MD5。
- v2026.07.18.1 (prod):Phase0 启动 — 部署前镜像备份 SOP + 版本号规范 + 引入测试基建。
- 旧方案:镜像曾用 AIFlow v2.0.1 / v2026.07.14 / v2026.07.14-nav;git 曾用 v2026.07.12-recover / v2026.07.12-ui-recover / v2026.07.13-recover。自 v2026.07.18.1 起统一为 vYYYY.MM.DD.N + prod/test,便于正式版推广后与测试版做环境切割。
- 历史功能:AI聊天(real API + tool-call function calling)、知识库文件存储 + AI 图谱、频道二维码、邮件/短信配置 + AI 生成、支付配置、平台驾驶舱(token 累计 + 审计/操作日志)、Agent 资源 API、evolution pipeline 自动恢复、全局默认模型、广告位图片上传 + AI 绘图、应用市场安装/卸载/搜索/去重、资产文件类型分类 + 在线预览。迁移 001-015。
v2026.07.19.5
@@ -0,0 +1,103 @@
"""调用可观测面板后端 API(媒体/LLM 调用归因)。
路由前缀: /app/api/admin/observability
数据来源: token_usage_logs(Phase 2.2 已加 trace_id 列,可经 X-Trace-Id 关联请求链路)。
"""
from datetime import datetime, timedelta
from fastapi import APIRouter, Query, Depends
from ...database import get_db
from ..auth import get_current_user
from sqlalchemy import text
from sqlalchemy.orm import Session
router = APIRouter()
@router.get("/summary")
async def observability_summary(
days: int = Query(30, description="统计天数"),
db: Session = Depends(get_db),
user=Depends(get_current_user),
):
"""媒体/LLM 调用归因:总览 + 按模型 + 按用户 + 每日趋势 + 最近调用(trace_id 下钻)。"""
since = datetime.now() - timedelta(days=days)
totals = db.execute(text("""
SELECT COUNT(*),
COALESCE(SUM(total_tokens), 0),
COALESCE(SUM(cost), 0),
COUNT(DISTINCT user_id)
FROM token_usage_logs WHERE created_at >= :since
"""), {"since": since}).fetchone()
by_model = db.execute(text("""
SELECT model_name, COUNT(*), COALESCE(SUM(total_tokens), 0), COALESCE(SUM(cost), 0)
FROM token_usage_logs WHERE created_at >= :since
GROUP BY model_name ORDER BY COALESCE(SUM(total_tokens), 0) DESC
"""), {"since": since}).fetchall()
by_user = db.execute(text("""
SELECT t.user_id, u.username, COUNT(*), COALESCE(SUM(t.total_tokens), 0), COALESCE(SUM(t.cost), 0)
FROM token_usage_logs t LEFT JOIN users u ON u.id = t.user_id
WHERE t.created_at >= :since
GROUP BY t.user_id, u.username ORDER BY COALESCE(SUM(t.total_tokens), 0) DESC LIMIT 20
"""), {"since": since}).fetchall()
daily = db.execute(text("""
SELECT DATE(created_at) AS d, COALESCE(SUM(total_tokens), 0), COALESCE(SUM(cost), 0), COUNT(*)
FROM token_usage_logs WHERE created_at >= :since
GROUP BY DATE(created_at) ORDER BY d
"""), {"since": since}).fetchall()
recent = db.execute(text("""
SELECT trace_id, model_name, total_tokens, cost, created_at
FROM token_usage_logs WHERE created_at >= :since
ORDER BY created_at DESC LIMIT 50
"""), {"since": since}).fetchall()
return {
"totals": {
"calls": int(totals[0] or 0),
"tokens": int(totals[1] or 0),
"cost": float(totals[2] or 0),
"users": int(totals[3] or 0),
},
"by_model": [
{
"model_name": m[0] or "unknown",
"calls": int(m[1] or 0),
"tokens": int(m[2] or 0),
"cost": float(m[3] or 0),
}
for m in by_model
],
"by_user": [
{
"user_id": u[0],
"username": u[1] or u[0],
"calls": int(u[2] or 0),
"tokens": int(u[3] or 0),
"cost": float(u[4] or 0),
}
for u in by_user
],
"daily": [
{
"date": str(d[0]),
"tokens": int(d[1] or 0),
"cost": float(d[2] or 0),
"calls": int(d[3] or 0),
}
for d in daily
],
"recent": [
{
"trace_id": r[0],
"model_name": r[1] or "unknown",
"tokens": int(r[2] or 0),
"cost": float(r[3] or 0),
"created_at": r[4].strftime("%Y-%m-%d %H:%M:%S") if r[4] else None,
}
for r in recent
],
}
+3 -1
View File
@@ -44,7 +44,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_agents import (_call_coze_agent, _call_dify_agent, _call_external_agent, _call_n8n_agent, _normalize_dify_chat_url)
from app.api.chat_context import (_build_system_prompt, _get_conversation_context, _ingest_answer_to_dify, _retrieve_kb_context)
from app.api.chat_context import (_build_system_prompt, _get_conversation_context, _ingest_answer_to_dify, _retrieve_kb_context, _build_chat_history, _maybe_compress_history)
from app.api.chat_knowledge import (_build_knowledge_graph)
from app.api.chat_mcp import (_list_enabled_mcp_servers, _resolve_active_mcp_servers, _resolve_market_mcp)
from app.api.chat_media import (_agnes_post_json, _friendly_media_err, _generate_media, _generate_media_task, _optimize_media_prompt, _write_media_token_log, detect_ambiguous_creative_intent, detect_media_intent)
@@ -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)
+161 -1
View File
@@ -20,7 +20,7 @@ from ..models import (
User, Conversation, Message, AIModel, ModelProvider, TokenUsageLog,
User, Conversation, Message, AIModel, ModelProvider, TokenUsageLog, ConversationSummary,
@@ -1230,3 +1230,163 @@ def _get_conversation_context(db: Session, conv_id: str) -> dict:
return context
# ===== Phase 2 上下文引擎升级:分层记忆 + 压缩 + 隔离 =====
logger = logging.getLogger(__name__)
def _estimate_tokens(text):
"""粗略 token 估算:中英文混排约 2 字符/token。"""
if not text:
return 0
return max(1, len(text) // 2)
def _build_chat_history(db, conv_id, max_tokens=6000, max_messages=40,
exclude_latest_user=True, include_summary=True):
"""会话层上下文引擎:预算感知的历史组装 + 可选压缩摘要前置。
- 若 conversation_summaries 存在摘要,以 system 角色「对话摘要」前置(压缩层)。
- 按 token 估算从新到旧累加,超预算即截断(隔离:不溢出上下文窗口)。
- exclude_latest_user=True 时丢弃最新一条 user 消息(当前轮次另行发送)。
"""
summary_block = None
if include_summary:
try:
summ = db.query(ConversationSummary).filter(
ConversationSummary.conversation_id == conv_id
).first()
if summ and summ.summary_text:
summary_block = {
"role": "system",
"content": "【对话历史摘要,请据此理解用户此前的意图与已确认结论】\n" + summ.summary_text,
}
except Exception:
pass
rows = db.query(Message).filter(
Message.conversation_id == conv_id,
Message.role.in_(["user", "assistant"]),
).order_by(desc(Message.created_at)).limit(max_messages).all()
if exclude_latest_user:
for i, m in enumerate(rows):
if m.role == "user":
rows = rows[:i] + rows[i + 1:]
break
kept = []
used = 0
for m in rows: # 新->旧
t = _estimate_tokens(m.content or "")
if kept and used + t > max_tokens:
break
kept.append({"role": m.role, "content": m.content})
used += t
kept.reverse() # 时间正序
out = []
if summary_block:
out.append(summary_block)
out.extend(kept)
return out
async def _summarize_via_llm(prompt):
"""调用默认文本模型生成对话摘要;失败返回 None(压缩为可选项,不阻塞)。"""
try:
from ..capability_registry import resolve_auto_model
from app.api.chat_utils import _proxy_for_url
sdb = SessionLocal()
try:
ai_model = resolve_auto_model("text", sdb)
if not ai_model:
return None
provider = sdb.query(ModelProvider).filter(
ModelProvider.id == ai_model.provider_id
).first()
api_key = ai_model.api_key or (provider.api_key if provider else None)
if not provider or not provider.api_base_url or not api_key:
return None
api_url = f"{provider.api_base_url.rstrip('/')}/v1/chat/completions"
body = {
"model": ai_model.name,
"messages": [
{"role": "system", "content": "你是专业的对话摘要助手,输出简洁中文摘要。"},
{"role": "user", "content": prompt},
],
"max_tokens": 512,
"temperature": 0.2,
}
async with httpx.AsyncClient(proxy=_proxy_for_url(api_url), trust_env=False, timeout=60) as client:
resp = await client.post(
api_url,
headers={"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"},
json=body,
)
if resp.status_code != 200:
return None
data = resp.json()
return data.get("choices", [{}])[0].get("message", {}).get("content", "").strip()
finally:
sdb.close()
except Exception as e:
logger.warning(f"[context] summarize llm failed: {e}")
return None
async def _maybe_compress_history(conv_id):
"""后台异步压缩:将受保护窗口外的旧消息摘要落库 conversation_summaries。
失败静默回滚,绝不阻塞用户本次响应(fire-and-forget)。
"""
sdb = None
try:
sdb = SessionLocal()
all_msgs = sdb.query(Message).filter(
Message.conversation_id == conv_id,
Message.role.in_(["user", "assistant"]),
).order_by(asc(Message.created_at)).all()
PROTECTED = 20
if len(all_msgs) <= PROTECTED:
return
compressible = all_msgs[: len(all_msgs) - PROTECTED]
if len(compressible) < 8:
return
chunk = compressible[:30]
text = "\n".join(f"{m.role}: {m.content}" for m in chunk)
existing = sdb.query(ConversationSummary).filter(
ConversationSummary.conversation_id == conv_id
).first()
prior = existing.summary_text if existing else ""
prompt = (
"请把以下对话片段浓缩为简明摘要,保留用户核心诉求、已确认结论、关键事实与待办。"
"若提供『已有摘要』,请合并而非重复。\n\n"
f"已有摘要:\n{prior}\n\n待摘要片段:\n{text}"
)
summary = await _summarize_via_llm(prompt)
if not summary:
return
if existing:
existing.summary_text = summary
existing.token_count = _estimate_tokens(summary)
existing.updated_at = datetime.now()
else:
sdb.add(ConversationSummary(
conversation_id=conv_id,
summary_text=summary,
token_count=_estimate_tokens(summary),
updated_at=datetime.now(),
))
sdb.commit()
except Exception as e:
logger.warning(f"[context] compress history failed conv={conv_id}: {e}")
if sdb is not None:
try:
sdb.rollback()
except Exception:
pass
finally:
if sdb is not None:
try:
sdb.close()
except Exception:
pass
+6 -2
View File
@@ -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,
+14 -43
View File
@@ -16,6 +16,7 @@ import httpx
import os
import logging
from ..database import get_db, SessionLocal
from app.core.observability import span, get_trace_id
from ..models import (
@@ -47,7 +48,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_agents import (_call_external_agent)
from app.api.chat_context import (_build_system_prompt, _get_conversation_context, _ingest_answer_to_dify, _retrieve_kb_context)
from app.api.chat_context import (_build_system_prompt, _get_conversation_context, _ingest_answer_to_dify, _retrieve_kb_context, _build_chat_history, _maybe_compress_history)
from app.api.chat_mcp import (_list_enabled_mcp_servers, _resolve_active_mcp_servers)
from app.api.chat_media import (_generate_media_task, detect_ambiguous_creative_intent, detect_media_intent)
from app.api.chat_skill import (_discover_installed_skills, _execute_skill_tool, _resolve_market_agents, _resolve_market_skills)
@@ -537,23 +538,7 @@ async def _send_message_core(
_hist_rows = db.query(Message).filter(
Message.conversation_id == conv_id
).order_by(asc(Message.created_at)).limit(10).all()
_history = [{"role": h.role, "content": h.content}
for h in _hist_rows if h.role in ("user", "assistant")]
_history = _build_chat_history(db, conv_id, max_tokens=4000, exclude_latest_user=True)
@@ -791,6 +776,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:
@@ -1064,31 +1050,8 @@ async def _send_message_core(
# 加载对话历史(最多10条)
history = db.query(Message).filter(
Message.conversation_id == conv_id
).order_by(asc(Message.created_at)).limit(10).all()
for h in history:
if h.role == "user" or h.role == "assistant":
messages.append({"role": h.role, "content": h.content})
# 分层上下文引擎:预算感知历史 + 压缩摘要(若有)
messages.extend(_build_chat_history(db, conv_id, max_tokens=6000, exclude_latest_user=True))
@@ -1192,6 +1155,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 +1344,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 +1355,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
@@ -1777,6 +1743,8 @@ async def _send_message_core(
conversation_id=conv_id,
trace_id=get_trace_id(),
@@ -1928,6 +1896,9 @@ async def _send_message_core(
# 后台异步压缩历史(分层记忆),不阻塞本次响应
asyncio.create_task(_maybe_compress_history(conv_id))
db.refresh(user_msg)
+2
View File
@@ -25,6 +25,7 @@ from sqlalchemy.orm import Session
from ..capability_registry import resolve_auto_model
from ..models import AIModel, ModelProvider, TokenUsageLog, gen_uuid
from app.core.observability import get_trace_id
logger = logging.getLogger(__name__)
@@ -95,6 +96,7 @@ def _log_media_usage(db: Session, user_id: str, ai_model, media_type: str, promp
user_id=user_id,
model_id=ai_model.id,
model_name=ai_model.name,
trace_id=get_trace_id(),
prompt_tokens=est,
completion_tokens=0,
total_tokens=est,
+8 -1
View File
@@ -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
+1
View File
@@ -0,0 +1 @@
# app/core package - 统一可观测基座 (Phase 2.1)
+174
View File
@@ -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",
]
+2 -1
View File
@@ -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()
+10
View File
@@ -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)
@@ -1124,3 +1125,12 @@ class SmsVerificationCode(Base):
used = Column(Boolean, default=False)
created_at = Column(DateTime, default=datetime.now)
class ConversationSummary(Base):
__tablename__ = "conversation_summaries"
conversation_id = Column(String(64), primary_key=True)
summary_text = Column(Text, nullable=False, default="")
token_count = Column(Integer, default=0)
updated_at = Column(DateTime, default=datetime.now, onupdate=datetime.now)
+5 -1
View File
@@ -17,7 +17,7 @@ sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from app.api import auth, chat, knowledge, market, token_usage, payment, profile, notifications, assets, evolution
from app.casdoor_auth import router as casdoor_router
from app.api.admin import dashboard, users, models, settings, payment_admin, services, membership_admin, ads, market_admin, logs, skills_admin, agents_admin, knowledge_admin, workflow_admin, storage_admin, automation_admin, mcp_admin, developer_admin, apps_admin, distro_admin, code_exec_admin, privacy_admin, im_bot_admin, promotion_admin, achievement_admin, tenant_admin, billing_admin, analytics_admin
from app.api.admin import dashboard, users, models, settings, payment_admin, services, membership_admin, ads, market_admin, logs, skills_admin, agents_admin, knowledge_admin, workflow_admin, storage_admin, automation_admin, mcp_admin, developer_admin, apps_admin, distro_admin, code_exec_admin, privacy_admin, im_bot_admin, promotion_admin, achievement_admin, tenant_admin, billing_admin, analytics_admin, observability_admin
from app.api.im_bot_user import router as im_bot_user_router
from app.api.im_bot_platform import router as im_bot_platform_router
from app.api.portal import user_portal
@@ -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(
@@ -236,6 +239,7 @@ app.include_router(promotion_admin.router, prefix="/app/api/admin/promotion", ta
app.include_router(tenant_admin.router, prefix="/app/api/admin/tenant", tags=["tenant"])
app.include_router(billing_admin.router, prefix="/app/api/admin/billing", tags=["billing"])
app.include_router(analytics_admin.router, prefix="/app/api/admin/analytics", tags=["analytics"])
app.include_router(observability_admin.router, prefix="/app/api/admin/observability", tags=["后台-调用可观测"])
app.include_router(user_portal.router, prefix="/app/api/portal", tags=["portal"])
app.include_router(achievement_admin.router, prefix="/app/api/admin/achievement", tags=["achievement"])
app.include_router(casdoor_router, prefix="/app/api/auth", tags=["Casdoor SSO"])
+114
View File
@@ -0,0 +1,114 @@
"""Phase 2 context-engine regression: _build_chat_history budget/layering/summary.
Pure unit test with a fake DB session — no network, no real model. Guards the
layered-memory + compression + isolation behaviour against regressions.
"""
from app.api.chat_context import _estimate_tokens, _build_chat_history
class _Msg:
def __init__(self, role, content, cid="c1"):
self.role = role
self.content = content
self.conversation_id = cid
class _Summary:
def __init__(self, text):
self.summary_text = text
class _Query:
def __init__(self, rows):
self._rows = list(rows)
def filter(self, *a, **k):
return self
def order_by(self, *a, **k):
# test pre-sorts newest-first to match desc(created_at)
return self
def limit(self, n=None):
return self
def all(self):
return self._rows
def first(self):
return self._rows[0] if self._rows else None
class _DB:
def __init__(self, messages, summary=None):
self._messages = messages
self._summary = summary
def query(self, model):
if getattr(model, "__name__", "") == "ConversationSummary":
return _Query([self._summary] if self._summary else [])
return _Query(self._messages)
def _newest_first(msgs):
# _build_chat_history expects desc(created_at) -> newest first.
# Test authors write the list already newest-first, so return as-is.
return list(msgs)
def test_estimate_tokens():
assert _estimate_tokens("") == 0
assert _estimate_tokens(None) == 0
assert _estimate_tokens("abcd") == 2 # 4//2
assert _estimate_tokens("a" * 5) == 2 # max(1, 5//2)=2
# just ensure positive monotonic
assert _estimate_tokens("x" * 100) == 50
def test_history_excludes_latest_user():
# newest-first: [current user, prev assistant, older user]
msgs = _newest_first([
_Msg("user", "current question"),
_Msg("assistant", "previous answer"),
_Msg("user", "older question"),
])
db = _DB(msgs)
out = _build_chat_history(db, "c1", max_tokens=10000, exclude_latest_user=True)
roles = [m["role"] for m in out]
# current user dropped; chronological order = [older user, previous assistant]
assert roles == ["user", "assistant"]
assert out[0]["content"] == "older question"
def test_summary_injected_as_system_first():
msgs = _newest_first([
_Msg("user", "hi"),
_Msg("assistant", "hello"),
])
db = _DB(msgs, summary=_Summary("用户想做图片生成"))
out = _build_chat_history(db, "c1", max_tokens=10000, exclude_latest_user=True)
assert out[0]["role"] == "system"
assert out[0]["content"].startswith("【对话历史摘要")
assert "用户想做图片生成" in out[0]["content"]
# after summary: chronological history (assistant then user 'hi' excluded? exclude_latest_user drops 'hi')
assert out[1]["role"] == "assistant"
def test_budget_truncation():
# 5 assistant messages each ~200 tokens; budget small -> truncated
msgs = _newest_first([_Msg("assistant", "x" * 400) for _ in range(5)])
db = _DB(msgs)
out = _build_chat_history(db, "c1", max_tokens=500, exclude_latest_user=False)
# each msg ~200 tokens; budget 500 -> keep newest ~2
assert len(out) <= 3
total = sum(_estimate_tokens(m["content"]) for m in out)
assert total <= 500 + 200 # within budget (+1 msg slack)
def test_no_summary_when_absent():
msgs = _newest_first([_Msg("user", "hi"), _Msg("assistant", "hello")])
db = _DB(msgs)
out = _build_chat_history(db, "c1", max_tokens=10000, exclude_latest_user=True)
# 'hi' excluded; only assistant remains, no system summary
assert all(m["role"] != "system" for m in out)
assert [m["role"] for m in out] == ["assistant"]
+2
View File
@@ -45,6 +45,7 @@ const Achievement = React.lazy(() => import('@/pages/admin/Achievement'));
const Tenant = React.lazy(() => import('@/pages/admin/Tenant'));
const Billing = React.lazy(() => import('@/pages/admin/Billing'));
const Analytics = React.lazy(() => import('@/pages/admin/Analytics'));
const Observability = React.lazy(() => import('@/pages/admin/Observability'));
const AdminKnowledge = React.lazy(() => import('@/pages/admin/Knowledge'));
const AdminWorkflow = React.lazy(() => import('@/pages/admin/Workflow'));
const AdminStorage = React.lazy(() => import('@/pages/admin/Storage'));
@@ -151,6 +152,7 @@ const App: React.FC = () => {
<Route path="tenant" element={<Tenant />} />
<Route path="billing" element={<Billing />} />
<Route path="analytics" element={<Analytics />} />
<Route path="observability" element={<Observability />} />
<Route path="knowledge" element={<AdminKnowledge />} />
<Route path="workflow" element={<AdminWorkflow />} />
<Route path="storage" element={<AdminStorage />} />
+1
View File
@@ -527,6 +527,7 @@ export const adminAPI = {
getRecentAudit: () => get<any[]>('/admin/dashboard/recent-audit'),
getDailyStats: (days?: number) => get<{ days: number; data: Array<{ date: string; tokens: number }> }>(`/admin/dashboard/daily-stats${days ? `?days=${days}` : ''}`),
getObservabilitySummary: (days?: number) => get<any>(`/admin/observability/summary${days ? `?days=${days}` : ''}`),
// 用户管理
@@ -97,6 +97,7 @@ const adminNavGroups: NavGroup[] = [
icon: 'developer',
items: [
{ path: '/admin/analytics', icon: 'analytics', label: '数据分析' },
{ path: '/admin/observability', icon: 'observability', label: '调用可观测' },
{ path: '/admin/workflow', icon: 'workflow', label: '工作流编排' },
{ path: '/admin/apps', icon: 'apps', label: 'AI 应用管理' },
{ path: '/admin/code-exec', icon: 'code', label: '代码执行沙箱' },
@@ -154,6 +155,7 @@ const adminIconPaths: Record<string, JSX.Element> = {
billing: <svg width="18" height="18" viewBox="0 0 24 24" fill="none" stroke="currentColor" strokeWidth="1.8" strokeLinecap="round" strokeLinejoin="round"><rect x="2" y="5" width="20" height="14" rx="2"/><line x1="2" y1="10" x2="22" y2="10"/><line x1="6" y1="15" x2="10" y2="15"/></svg>,
analytics: <svg width="18" height="18" viewBox="0 0 24 24" fill="none" stroke="currentColor" strokeWidth="1.8" strokeLinecap="round" strokeLinejoin="round"><line x1="18" y1="20" x2="18" y2="10"/><line x1="12" y1="20" x2="12" y2="4"/><line x1="6" y1="20" x2="6" y2="14"/><line x1="3" y1="20" x2="21" y2="20"/></svg>,
observability: <svg width="18" height="18" viewBox="0 0 24 24" fill="none" stroke="currentColor" strokeWidth="1.8" strokeLinecap="round" strokeLinejoin="round"><path d="M3 12h4l3 8 4-16 3 8h4"/></svg>,
imbot: <svg width="18" height="18" viewBox="0 0 24 24" fill="none" stroke="currentColor" strokeWidth="1.8" strokeLinecap="round" strokeLinejoin="round"><path d="M21 11.5a8.38 8.38 0 0 1-.9 3.8 8.5 8.5 0 0 1-7.6 4.7 8.38 8.38 0 0 1-3.8-.9L3 21l1.9-5.7a8.38 8.38 0 0 1-.9-3.8 8.5 8.5 0 0 1 4.7-7.6 8.38 8.38 0 0 1 3.8-.9h.5a8.48 8.48 0 0 1 8 8v.5z"/></svg>,
+165
View File
@@ -0,0 +1,165 @@
import { useEffect, useState } from 'react';
import { adminAPI } from '@/api';
interface Summary {
totals: { calls: number; tokens: number; cost: number; users: number };
by_model: { model_name: string; calls: number; tokens: number; cost: number }[];
by_user: { user_id: string; username: string; calls: number; tokens: number; cost: number }[];
daily: { date: string; tokens: number; cost: number; calls: number }[];
recent: { trace_id: string | null; model_name: string; tokens: number; cost: number; created_at: string | null }[];
}
const ACCENT = '#5B5FED';
const ACCENT_HOVER = '#4A4ECC';
export default function Observability() {
const [days, setDays] = useState(30);
const [data, setData] = useState<Summary | null>(null);
const [loading, setLoading] = useState(false);
const [err, setErr] = useState('');
useEffect(() => {
let alive = true;
setLoading(true);
setErr('');
adminAPI.getObservabilitySummary(days)
.then((d: Summary) => { if (alive) setData(d); })
.catch((e: any) => { if (alive) setErr(e?.message || '加载失败'); })
.finally(() => { if (alive) setLoading(false); });
return () => { alive = false; };
}, [days]);
const maxTokens = data ? Math.max(1, ...data.daily.map(d => d.tokens)) : 1;
const maxModel = data ? Math.max(1, ...data.by_model.map(m => m.tokens)) : 1;
const fmt = (n: number) => n.toLocaleString('zh-CN');
const fmtCost = (n: number) => `¥${n.toFixed(4)}`;
const card: React.CSSProperties = {
background: '#fff', border: '1px solid #eceef5', borderRadius: 16,
padding: 20, boxShadow: '0 1px 3px rgba(20,30,60,0.04)', marginBottom: 18,
};
const stat: React.CSSProperties = {
flex: 1, minWidth: 150, background: '#fff', border: '1px solid #eceef5',
borderRadius: 16, padding: 18, boxShadow: '0 1px 3px rgba(20,30,60,0.04)',
};
return (
<div style={{ padding: 24, maxWidth: 1160, margin: '0 auto', color: '#1f2330' }}>
<div style={{ display: 'flex', alignItems: 'center', justifyContent: 'space-between', marginBottom: 20, flexWrap: 'wrap', gap: 12 }}>
<div>
<h2 style={{ margin: 0, fontSize: 22, fontWeight: 700 }}>调用可观测</h2>
<div style={{ color: '#8a90a2', fontSize: 13, marginTop: 4 }}>
媒体 / LLM 调用归因 · 基于 trace_id 关联请求链路(Phase 2.2)
</div>
</div>
<div style={{ display: 'flex', gap: 8 }}>
{[7, 30, 90].map(d => (
<button key={d} onClick={() => setDays(d)}
style={{
padding: '7px 14px', borderRadius: 10, cursor: 'pointer', fontSize: 13,
border: `1px solid ${days === d ? ACCENT : '#e2e5ee'}`,
background: days === d ? ACCENT : '#fff',
color: days === d ? '#fff' : '#5b6172', fontWeight: 600,
transition: 'all .15s',
}}>
{d} 天
</button>
))}
</div>
</div>
{err && <div style={{ ...card, borderColor: '#f3c2c2', color: '#c0392b' }}>{err}</div>}
{loading && !data && <div style={{ color: '#8a90a2' }}>加载中…</div>}
{data && (
<>
<div style={{ display: 'flex', gap: 14, flexWrap: 'wrap', marginBottom: 18 }}>
<div style={stat}><div style={{ color: '#8a90a2', fontSize: 12 }}>调用次数</div><div style={{ fontSize: 24, fontWeight: 700, marginTop: 6 }}>{fmt(data.totals.calls)}</div></div>
<div style={stat}><div style={{ color: '#8a90a2', fontSize: 12 }}>Token 消耗</div><div style={{ fontSize: 24, fontWeight: 700, marginTop: 6 }}>{fmt(data.totals.tokens)}</div></div>
<div style={stat}><div style={{ color: '#8a90a2', fontSize: 12 }}>成本</div><div style={{ fontSize: 24, fontWeight: 700, marginTop: 6, color: ACCENT }}>{fmtCost(data.totals.cost)}</div></div>
<div style={stat}><div style={{ color: '#8a90a2', fontSize: 12 }}>活跃用户</div><div style={{ fontSize: 24, fontWeight: 700, marginTop: 6 }}>{fmt(data.totals.users)}</div></div>
</div>
<div style={card}>
<h3 style={{ margin: '0 0 14px', fontSize: 16 }}>按模型</h3>
<div style={{ display: 'flex', flexDirection: 'column', gap: 10 }}>
{data.by_model.length === 0 && <div style={{ color: '#8a90a2', fontSize: 13 }}>暂无数据</div>}
{data.by_model.map(m => (
<div key={m.model_name} style={{ display: 'flex', alignItems: 'center', gap: 12 }}>
<div style={{ width: 150, fontSize: 13, fontWeight: 600, overflow: 'hidden', textOverflow: 'ellipsis', whiteSpace: 'nowrap' }}>{m.model_name}</div>
<div style={{ flex: 1, height: 10, background: '#f1f3f9', borderRadius: 6, overflow: 'hidden' }}>
<div style={{ width: `${(m.tokens / maxModel) * 100}%`, height: '100%', background: ACCENT, borderRadius: 6 }} />
</div>
<div style={{ width: 90, textAlign: 'right', fontSize: 13, color: '#5b6172' }}>{fmt(m.tokens)}</div>
<div style={{ width: 90, textAlign: 'right', fontSize: 13, color: ACCENT }}>{fmtCost(m.cost)}</div>
</div>
))}
</div>
</div>
<div style={card}>
<h3 style={{ margin: '0 0 14px', fontSize: 16 }}>每日趋势(Token)</h3>
<div style={{ display: 'flex', alignItems: 'flex-end', gap: 6, height: 140, overflowX: 'auto', paddingBottom: 4 }}>
{data.daily.length === 0 && <div style={{ color: '#8a90a2', fontSize: 13 }}>暂无数据</div>}
{data.daily.map(d => (
<div key={d.date} title={`${d.date} · ${fmt(d.tokens)} tokens · ${fmtCost(d.cost)}`}
style={{ display: 'flex', flexDirection: 'column', alignItems: 'center', gap: 4, minWidth: 22 }}>
<div style={{ width: 16, height: `${(d.tokens / maxTokens) * 110}px`, background: ACCENT, borderRadius: '4px 4px 0 0', opacity: 0.85 }} />
<div style={{ fontSize: 9, color: '#aab', transform: 'rotate(-45deg)', whiteSpace: 'nowrap' }}>{d.date.slice(5)}</div>
</div>
))}
</div>
</div>
<div style={{ display: 'flex', gap: 18, flexWrap: 'wrap' }}>
<div style={{ ...card, flex: 1, minWidth: 320 }}>
<h3 style={{ margin: '0 0 14px', fontSize: 16 }}>按用户(Top 20)</h3>
<table style={{ width: '100%', borderCollapse: 'collapse', fontSize: 13 }}>
<thead><tr style={{ color: '#8a90a2', textAlign: 'left' }}>
<th style={th}>用户</th><th style={thR}>调用</th><th style={thR}>Token</th><th style={thR}>成本</th>
</tr></thead>
<tbody>
{data.by_user.map(u => (
<tr key={u.user_id} style={{ borderTop: '1px solid #f1f3f9' }}>
<td style={td}>{u.username}</td>
<td style={tdR}>{fmt(u.calls)}</td>
<td style={tdR}>{fmt(u.tokens)}</td>
<td style={{ ...tdR, color: ACCENT }}>{fmtCost(u.cost)}</td>
</tr>
))}
{data.by_user.length === 0 && <tr><td colSpan={4} style={{ ...td, color: '#8a90a2' }}>暂无数据</td></tr>}
</tbody>
</table>
</div>
</div>
<div style={card}>
<h3 style={{ margin: '0 0 14px', fontSize: 16 }}>最近调用(trace_id 可下钻)</h3>
<table style={{ width: '100%', borderCollapse: 'collapse', fontSize: 13 }}>
<thead><tr style={{ color: '#8a90a2', textAlign: 'left' }}>
<th style={th}>Trace ID</th><th style={th}>模型</th><th style={thR}>Token</th><th style={thR}>成本</th><th style={thR}>时间</th>
</tr></thead>
<tbody>
{data.recent.map((r, i) => (
<tr key={(r.trace_id || '') + i} style={{ borderTop: '1px solid #f1f3f9' }}>
<td style={{ ...td, fontFamily: 'monospace', fontSize: 12, maxWidth: 200, overflow: 'hidden', textOverflow: 'ellipsis', whiteSpace: 'nowrap' }}>{r.trace_id || '—'}</td>
<td style={td}>{r.model_name}</td>
<td style={tdR}>{fmt(r.tokens)}</td>
<td style={{ ...tdR, color: ACCENT }}>{fmtCost(r.cost)}</td>
<td style={tdR}>{r.created_at || '—'}</td>
</tr>
))}
{data.recent.length === 0 && <tr><td colSpan={5} style={{ ...td, color: '#8a90a2' }}>暂无数据</td></tr>}
</tbody>
</table>
</div>
</>
)}
</div>
);
}
const th: React.CSSProperties = { padding: '8px 6px', fontWeight: 600, fontSize: 12 };
const thR: React.CSSProperties = { ...th, textAlign: 'right' };
const td: React.CSSProperties = { padding: '8px 6px' };
const tdR: React.CSSProperties = { ...td, textAlign: 'right', color: '#5b6172' };
@@ -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;
@@ -0,0 +1,7 @@
CREATE TABLE IF NOT EXISTS conversation_summaries (
conversation_id VARCHAR(64) NOT NULL,
summary_text TEXT NULL,
token_count INT NULL DEFAULT 0,
updated_at DATETIME(3) NULL,
PRIMARY KEY (conversation_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;