diff --git a/VERSION b/VERSION index 3dec58e..e2dac30 100644 --- a/VERSION +++ b/VERSION @@ -3,11 +3,11 @@ # 版本号格式: vYYYY.MM.DD.N (年.月.日.当日递增序号，从 1 开始) # 环境标签: prod = 正式版 | test = 测试版 -VERSION=v2026.07.19.1 +VERSION=v2026.07.19.2 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-002 ## 本版本变更 (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 +16,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 时区格式化）。 diff --git a/backend/python/app/api/chat.py b/backend/python/app/api/chat.py index 7b9562a..2d053bc 100644 --- a/backend/python/app/api/chat.py +++ b/backend/python/app/api/chat.py @@ -1,1383 +1,1385 @@ -from fastapi import APIRouter, Depends, HTTPException, Query, Response -from sqlalchemy.orm import Session -from sqlalchemy import and_, desc, asc, func -from pydantic import BaseModel, Field -from typing import Optional -import uuid -from datetime import datetime, timezone, timedelta -import re -import json -import asyncio -import math -import httpx -import os -import logging -from ..database import get_db, SessionLocal -from ..models import ( - - - - User, Conversation, Message, AIModel, ModelProvider, TokenUsageLog, - - - - KnowledgeBase, KnowledgeDocument, KnowledgeDocumentChunk, EvolutionSkill, - - - - EvolutionTrajectory, MarketApp, AppPurchase, AppInstallation, - - - - MCPServer, - - - - gen_uuid - - - -) -from ..auth import get_current_user -from ..platform_knowledge import get_platform_knowledge -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_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) -from app.api.chat_skill import (_bm25_search, _discover_installed_skills, _execute_skill_tool, _parse_skill_content, _resolve_market_agents, _resolve_market_skills) -from app.api.chat_utils import (CST_TZ, CreateConversationRequest, SendMessageRequest, UpdateConversationRequest, _HK_PROXY, _extract_text_from_json, _html_escape, _outbound_proxy, _proxy_for_url, cst_now, logger) - -""" - - - -增强版 Chat API — Phase 2.5 - - - -包含: Skill ID统一(MarketApp映射), 真实Skill执行, Agent工作流注入, - - - - 多轮上下文保持, BM25 KB检索, Function Calling循环增强 - - - -""" - -from app.api.chat_service import _send_message_core - -router = APIRouter() - -_MEDIA_RELAY_ALLOWED = ("agnes-ai.com", "agnes-ai.space", "qihuyun.com") - -@router.get("/media/relay") - - - -async def media_relay(url: str = Query(...)): - - - - """中转 AGNES 媒体(图片/视频)到平台域名，供微信/iLink 内联渲染。 - - - - - - - - 仅允许白名单域名；否则 400。返回的媒体流保持原 content-type。 - - - - """ - - - - import urllib.parse as _up - - - - host = (_up.urlparse(url).hostname or "").lower() - - - - if not any(allowed in host for allowed in _MEDIA_RELAY_ALLOWED): - - - - raise HTTPException(status_code=400, detail="unsupported media host") - - - - try: - - - - async with httpx.AsyncClient(timeout=180, proxy=_proxy_for_url(url), trust_env=False) as client: - - - - r = await client.get(url) - - - - if r.status_code != 200: - - - - raise HTTPException(status_code=r.status_code, detail="upstream error") - - - - content_type = r.headers.get("content-type", "application/octet-stream") - - - - data = await r.aread() - - - - return Response(content=data, media_type=content_type) - - - - except HTTPException: - - - - raise - - - - except Exception as e: - - - - raise HTTPException(status_code=502, detail=f"relay failed: {e}") - -@router.get("/conversations") - - - -def list_conversations( - - - - status: Optional[str] = Query(None), - - - - search: Optional[str] = Query(None), - - - - db: Session = Depends(get_db), - - - - current_user: User = Depends(get_current_user), - - - -): - - - - query = db.query(Conversation).filter(Conversation.user_id == current_user.id) - - - - - - - - if status: - - - - query = query.filter(Conversation.status == status) - - - - else: - - - - query = query.filter(Conversation.status != "archived") - - - - - - - - if search: - - - - query = query.filter(Conversation.title.ilike(f"%{search}%")) - - - - - - - - query = query.order_by(desc(Conversation.pinned), desc(Conversation.updated_at)) - - - - - - - - conversations = query.all() - - - - result = [] - - - - for conv in conversations: - - - - result.append({ - - - - "id": conv.id, - - - - "title": conv.title, - - - - "model_id": conv.model_id, - - - - "status": conv.status, - - - - "pinned": conv.pinned, - - - - "knowledge_graph_count": conv.knowledge_graph_count or 0, - - - - "source": getattr(conv, "source", None), - - - - "created_at": str(conv.created_at) if conv.created_at else None, - - - - "updated_at": str(conv.updated_at) if conv.updated_at else None, - - - - }) - - - - - - - - return {"code": 0, "data": result, "message": "ok"} - -@router.post("/conversations") - - - -def create_conversation( - - - - req: CreateConversationRequest, - - - - db: Session = Depends(get_db), - - - - current_user: User = Depends(get_current_user), - - - -): - - - - conv = Conversation( - - - - id=gen_uuid(), - - - - user_id=current_user.id, - - - - title=req.title, - - - - model_id=req.model_id, - - - - status="active", - - - - pinned=False, - - - - knowledge_graph_count=0, - - - - created_at=cst_now(), - - - - updated_at=cst_now(), - - - - ) - - - - db.add(conv) - - - - db.commit() - - - - db.refresh(conv) - - - - - - - - return { - - - - "code": 0, - - - - "data": { - - - - "id": conv.id, - - - - "title": conv.title, - - - - "model_id": conv.model_id, - - - - "status": conv.status, - - - - "pinned": conv.pinned, - - - - "knowledge_graph_count": conv.knowledge_graph_count, - - - - "created_at": str(conv.created_at) if conv.created_at else None, - - - - "updated_at": str(conv.updated_at) if conv.updated_at else None, - - - - }, - - - - "message": "创建成功", - - - - } - -@router.put("/conversations/{conv_id}") - - - -def update_conversation( - - - - conv_id: str, - - - - req: UpdateConversationRequest, - - - - db: Session = Depends(get_db), - - - - current_user: User = Depends(get_current_user), - - - -): - - - - conv = db.query(Conversation).filter( - - - - and_(Conversation.id == conv_id, Conversation.user_id == current_user.id) - - - - ).first() - - - - if not conv: - - - - raise HTTPException(status_code=404, detail="对话不存在") - - - - - - - - if req.title is not None: - - - - conv.title = req.title - - - - if req.status is not None: - - - - conv.status = req.status - - - - if req.pinned is not None: - - - - conv.pinned = req.pinned - - - - # Stage B.2: 保存agent/skill/kb选择到对话上下文 - - - - # (存储在第一条system消息的knowledge_ids/agent_ids/skill_ids字段中) - - - - conv.updated_at = cst_now() - - - - - - - - db.commit() - - - - db.refresh(conv) - - - - - - - - return {"code": 0, "data": { - - - - "id": conv.id, "title": conv.title, "status": conv.status, "pinned": conv.pinned, - - - - "updated_at": str(conv.updated_at), - - - - }, "message": "更新成功"} - -@router.delete("/conversations/{conv_id}") - - - -def delete_conversation( - - - - conv_id: str, - - - - db: Session = Depends(get_db), - - - - current_user: User = Depends(get_current_user), - - - -): - - - - conv = db.query(Conversation).filter( - - - - and_(Conversation.id == conv_id, Conversation.user_id == current_user.id) - - - - ).first() - - - - if not conv: - - - - raise HTTPException(status_code=404, detail="对话不存在") - - - - - - - - conv.status = "archived" - - - - conv.updated_at = cst_now() - - - - db.commit() - - - - - - - - return {"code": 0, "message": "删除成功"} - -@router.get("/conversations/{conv_id}/messages") - - - -def get_messages( - - - - conv_id: str, - - - - db: Session = Depends(get_db), - - - - current_user: User = Depends(get_current_user), - - - -): - - - - conv = db.query(Conversation).filter( - - - - and_(Conversation.id == conv_id, Conversation.user_id == current_user.id) - - - - ).first() - - - - if not conv: - - - - raise HTTPException(status_code=404, detail="对话不存在") - - - - - - - - messages = ( - - - - db.query(Message) - - - - .filter(Message.conversation_id == conv_id) - - - - .order_by(asc(Message.created_at), asc(Message.id)) - - - - .all() - - - - ) - - - - - - - - result = [] - - - - for msg in messages: - - - - result.append({ - - - - "id": msg.id, - - - - "conversation_id": msg.conversation_id, - - - - "role": msg.role, - - - - "content": msg.content, - - - - "token_used": msg.token_used or 0, - - - - "model_id": msg.model_id, - - - - "used_skills": msg.used_skills or [], - "used_mcp": msg.used_mcp or [], - - - - "created_at": str(msg.created_at) if msg.created_at else None, - - - - }) - - - - - - - - return {"code": 0, "data": result, "message": "ok"} - -@router.post("/conversations/{conv_id}/messages") - - - -async def send_message( - - - - conv_id: str, - - - - req: SendMessageRequest, - - - - db: Session = Depends(get_db), - - - - current_user: User = Depends(get_current_user), - - - -): - return await _send_message_core(conv_id, req, db, current_user) - - -@router.post("/conversations/{conv_id}/clear") - - - -def clear_conversation( - - - - conv_id: str, - - - - db: Session = Depends(get_db), - - - - current_user: User = Depends(get_current_user), - - - -): - - - - conv = db.query(Conversation).filter( - - - - and_(Conversation.id == conv_id, Conversation.user_id == current_user.id) - - - - ).first() - - - - if not conv: - - - - raise HTTPException(status_code=404, detail="对话不存在") - - - - - - - - db.query(Message).filter( - - - - Message.conversation_id == conv_id - - - - ).delete() - - - - - - - - conv.knowledge_graph_count = 0 - - - - conv.updated_at = cst_now() - - - - db.commit() - - - - - - - - return {"code": 0, "message": "对话上下文已清空"} - -@router.post("/conversations/{conv_id}/to-knowledge-graph") - - - -def convert_to_knowledge_graph( - - - - conv_id: str, - - - - db: Session = Depends(get_db), - - - - current_user: User = Depends(get_current_user), - - - -): - - - - """将对话内容转换为知识图谱（幂等：同一对话仅生成一次知识库）""" - - - - conv = db.query(Conversation).filter( - - - - and_(Conversation.id == conv_id, Conversation.user_id == current_user.id) - - - - ).first() - - - - if not conv: - - - - raise HTTPException(status_code=404, detail="对话不存在") - - - - - - - - messages = ( - - - - db.query(Message) - - - - .filter(Message.conversation_id == conv_id) - - - - .order_by(asc(Message.created_at), asc(Message.id)) - - - - .all() - - - - ) - - - - - - - - if not messages: - - - - raise HTTPException(status_code=400, detail="对话中没有消息, 无法生成知识图谱") - - - - - - - - # 幂等：若该对话已生成过知识图谱且知识库仍存在，则直接刷新返回 - - - - if conv.knowledge_graph_base_id: - - - - existing_kb = db.query(KnowledgeBase).filter( - - - - KnowledgeBase.id == conv.knowledge_graph_base_id - - - - ).first() - - - - if existing_kb: - - - - graph_data = _build_knowledge_graph(db, existing_kb) - - - - conv.knowledge_graph_count = 1 - - - - conv.updated_at = cst_now() - - - - db.commit() - - - - return { - - - - "code": 0, - - - - "data": { - - - - "knowledge_base_id": existing_kb.id, - - - - "knowledge_base_name": existing_kb.name, - - - - "graph": graph_data, - - - - "knowledge_graph_count": 1, - - - - "already_exists": True, - - - - }, - - - - "message": "知识图谱已存在, 已刷新展示", - - - - } - - - - - - - - kb_name = f"对话图谱-{conv.title}" - - - - kb = KnowledgeBase( - - - - id=gen_uuid(), - - - - user_id=current_user.id, - - - - name=kb_name, - - - - description=f"从对话「{conv.title}」自动生成的知识图谱", - - - - directory_path="/", - - - - document_count=0, - - - - created_at=cst_now(), - - - - updated_at=cst_now(), - - - - ) - - - - db.add(kb) - - - - db.flush() - - - - - - - - doc_count = 0 - - - - for idx, msg in enumerate(messages): - - - - role_label = "用户" if msg.role == "user" else "AI" - - - - preview = re.sub(r'[\\n\\r\\t]+', ' ', msg.content[:40]).strip() - - - - filename = f"msg_{idx+1:03d}_{role_label}_{preview}.txt" - - - - - - - - doc = KnowledgeDocument( - - - - id=gen_uuid(), - - - - knowledge_base_id=kb.id, - - - - filename=filename, - - - - file_type="txt", - - - - file_size=len(msg.content.encode('utf-8')), - - - - content=msg.content, - - - - status="completed", - - - - chunk_count=max(1, len(msg.content) // 500 + 1), - - - - created_at=cst_now(), - - - - updated_at=cst_now(), - - - - ) - - - - db.add(doc) - - - - doc_count += 1 - - - - - - - - kb.document_count = doc_count - - - - db.flush() - - - - - - - - # 构建知识图谱节点和边 - - - - graph_data = _build_knowledge_graph(db, kb) - - - - conv.knowledge_graph_base_id = kb.id - - - - conv.knowledge_graph_count = 1 - - - - conv.updated_at = cst_now() - - - - - - - - db.commit() - - - - - - - - LogService.log_operation( - - - - user_id=current_user.id, username=current_user.username, action="convert_to_graph", - - - - target="conversation", target_id=conv_id, - - - - detail=f"将对话 {conv.title} 转换为知识图谱 {kb.name}", - - - - ) - - - - - - - - return { - - - - "code": 0, - - - - "data": { - - - - "knowledge_base_id": kb.id, - - - - "knowledge_base_name": kb.name, - - - - "graph": graph_data, - - - - "knowledge_graph_count": 1, - - - - "already_exists": False, - - - - }, - - - - "message": f"知识图谱构建完成", - - - - } - -@router.get("/models") - - - -def get_models(db: Session = Depends(get_db)): - - - - models = db.query(AIModel).filter(AIModel.is_enabled == True).all() - - - - result = [] - - - - for m in models: - - - - provider = db.query(ModelProvider).filter(ModelProvider.id == m.provider_id).first() - - - - result.append({ - - - - "id": m.id, - - - - "name": m.name, - - - - "display_name": getattr(m, "display_name", None) or m.name, - - - - "provider_id": m.provider_id, - - - - "provider_name": provider.name if provider else "", - - - - "capabilities": m.capabilities or [], - - - - "status": m.status, - - - - "price_per_million_tokens": float(m.price_per_million_tokens or 0), - - - - "is_enabled": m.is_enabled, - - - - "is_default": bool(m.is_default) if hasattr(m, 'is_default') else False, - - - - "total_calls": m.total_calls or 0, - - - - "total_tokens": m.total_tokens or 0, - - - - "created_at": str(m.created_at) if m.created_at else None, - - - - }) - - - - result.sort(key=lambda x: (0 if x["is_default"] else 1, x["name"])) - - - - return {"code": 0, "data": result, "message": "ok"} +from fastapi import APIRouter, Depends, HTTPException, Query, Response +from sqlalchemy.orm import Session +from sqlalchemy import and_, desc, asc, func +from pydantic import BaseModel, Field +from typing import Optional +import uuid +from datetime import datetime, timezone, timedelta +import re +import json +import asyncio +import math +import httpx +import os +import logging +from ..database import get_db, SessionLocal +from ..models import ( + + + + User, Conversation, Message, AIModel, ModelProvider, TokenUsageLog, + + + + KnowledgeBase, KnowledgeDocument, KnowledgeDocumentChunk, EvolutionSkill, + + + + EvolutionTrajectory, MarketApp, AppPurchase, AppInstallation, + + + + MCPServer, + + + + gen_uuid + + + +) +from ..auth import get_current_user +from ..platform_knowledge import get_platform_knowledge +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_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) +from app.api.chat_skill import (_bm25_search, _discover_installed_skills, _execute_skill_tool, _parse_skill_content, _resolve_market_agents, _resolve_market_skills) +from app.api.chat_utils import (CST_TZ, CreateConversationRequest, SendMessageRequest, UpdateConversationRequest, _HK_PROXY, _extract_text_from_json, _html_escape, _outbound_proxy, _proxy_for_url, cst_now, logger) + +""" + + + +增强版 Chat API — Phase 2.5 + + + +包含: Skill ID统一(MarketApp映射), 真实Skill执行, Agent工作流注入, + + + + 多轮上下文保持, BM25 KB检索, Function Calling循环增强 + + + +""" + +from app.api.chat_service import _send_message_core + +router = APIRouter() + +_MEDIA_RELAY_ALLOWED = ("agnes-ai.com", "agnes-ai.space", "qihuyun.com") + +@router.get("/media/relay") + + + +async def media_relay(url: str = Query(...)): + + + + """中转 AGNES 媒体(图片/视频)到平台域名，供微信/iLink 内联渲染。 + + + + + + + + 仅允许白名单域名；否则 400。返回的媒体流保持原 content-type。 + + + + """ + + + + import urllib.parse as _up + + + + host = (_up.urlparse(url).hostname or "").lower() + + + + if not any(allowed in host for allowed in _MEDIA_RELAY_ALLOWED): + + + + raise HTTPException(status_code=400, detail="unsupported media host") + + + + try: + + + + async with httpx.AsyncClient(timeout=180, proxy=_proxy_for_url(url), trust_env=False) as client: + + + + r = await client.get(url) + + + + if r.status_code != 200: + + + + raise HTTPException(status_code=r.status_code, detail="upstream error") + + + + content_type = r.headers.get("content-type", "application/octet-stream") + + + + data = await r.aread() + + + + return Response(content=data, media_type=content_type) + + + + except HTTPException: + + + + raise + + + + except Exception as e: + + + + raise HTTPException(status_code=502, detail=f"relay failed: {e}") + +@router.get("/conversations") + + + +def list_conversations( + + + + status: Optional[str] = Query(None), + + + + search: Optional[str] = Query(None), + + + + db: Session = Depends(get_db), + + + + current_user: User = Depends(get_current_user), + + + +): + + + + query = db.query(Conversation).filter(Conversation.user_id == current_user.id) + + + + + + + + if status: + + + + query = query.filter(Conversation.status == status) + + + + else: + + + + query = query.filter(Conversation.status != "archived") + + + + + + + + if search: + + + + query = query.filter(Conversation.title.ilike(f"%{search}%")) + + + + + + + + query = query.order_by(desc(Conversation.pinned), desc(Conversation.updated_at)) + + + + + + + + conversations = query.all() + + + + result = [] + + + + for conv in conversations: + + + + result.append({ + + + + "id": conv.id, + + + + "title": conv.title, + + + + "model_id": conv.model_id, + + + + "status": conv.status, + + + + "pinned": conv.pinned, + + + + "knowledge_graph_count": conv.knowledge_graph_count or 0, + + + + "source": getattr(conv, "source", None), + + + + "created_at": str(conv.created_at) if conv.created_at else None, + + + + "updated_at": str(conv.updated_at) if conv.updated_at else None, + + + + }) + + + + + + + + return {"code": 0, "data": result, "message": "ok"} + +@router.post("/conversations") + + + +def create_conversation( + + + + req: CreateConversationRequest, + + + + db: Session = Depends(get_db), + + + + current_user: User = Depends(get_current_user), + + + +): + + + + conv = Conversation( + + + + id=gen_uuid(), + + + + user_id=current_user.id, + + + + title=req.title, + + + + model_id=req.model_id, + + + + status="active", + + + + pinned=False, + + + + knowledge_graph_count=0, + + + + created_at=cst_now(), + + + + updated_at=cst_now(), + + + + ) + + + + db.add(conv) + + + + db.commit() + + + + db.refresh(conv) + + + + + + + + return { + + + + "code": 0, + + + + "data": { + + + + "id": conv.id, + + + + "title": conv.title, + + + + "model_id": conv.model_id, + + + + "status": conv.status, + + + + "pinned": conv.pinned, + + + + "knowledge_graph_count": conv.knowledge_graph_count, + + + + "created_at": str(conv.created_at) if conv.created_at else None, + + + + "updated_at": str(conv.updated_at) if conv.updated_at else None, + + + + }, + + + + "message": "创建成功", + + + + } + +@router.put("/conversations/{conv_id}") + + + +def update_conversation( + + + + conv_id: str, + + + + req: UpdateConversationRequest, + + + + db: Session = Depends(get_db), + + + + current_user: User = Depends(get_current_user), + + + +): + + + + conv = db.query(Conversation).filter( + + + + and_(Conversation.id == conv_id, Conversation.user_id == current_user.id) + + + + ).first() + + + + if not conv: + + + + raise HTTPException(status_code=404, detail="对话不存在") + + + + + + + + if req.title is not None: + + + + conv.title = req.title + + + + if req.status is not None: + + + + conv.status = req.status + + + + if req.pinned is not None: + + + + conv.pinned = req.pinned + + + + # Stage B.2: 保存agent/skill/kb选择到对话上下文 + + + + # (存储在第一条system消息的knowledge_ids/agent_ids/skill_ids字段中) + + + + conv.updated_at = cst_now() + + + + + + + + db.commit() + + + + db.refresh(conv) + + + + + + + + return {"code": 0, "data": { + + + + "id": conv.id, "title": conv.title, "status": conv.status, "pinned": conv.pinned, + + + + "updated_at": str(conv.updated_at), + + + + }, "message": "更新成功"} + +@router.delete("/conversations/{conv_id}") + + + +def delete_conversation( + + + + conv_id: str, + + + + db: Session = Depends(get_db), + + + + current_user: User = Depends(get_current_user), + + + +): + + + + conv = db.query(Conversation).filter( + + + + and_(Conversation.id == conv_id, Conversation.user_id == current_user.id) + + + + ).first() + + + + if not conv: + + + + raise HTTPException(status_code=404, detail="对话不存在") + + + + + + + + conv.status = "archived" + + + + conv.updated_at = cst_now() + + + + db.commit() + + + + + + + + return {"code": 0, "message": "删除成功"} + +@router.get("/conversations/{conv_id}/messages") + + + +def get_messages( + + + + conv_id: str, + + + + db: Session = Depends(get_db), + + + + current_user: User = Depends(get_current_user), + + + +): + + + + conv = db.query(Conversation).filter( + + + + and_(Conversation.id == conv_id, Conversation.user_id == current_user.id) + + + + ).first() + + + + if not conv: + + + + raise HTTPException(status_code=404, detail="对话不存在") + + + + + + + + messages = ( + + + + db.query(Message) + + + + .filter(Message.conversation_id == conv_id) + + + + .order_by(asc(Message.created_at), asc(Message.id)) + + + + .all() + + + + ) + + + + + + + + result = [] + + + + for msg in messages: + + + + result.append({ + + + + "id": msg.id, + + + + "conversation_id": msg.conversation_id, + + + + "role": msg.role, + + + + "content": msg.content, + + + + "token_used": msg.token_used or 0, + + + + "model_id": msg.model_id, + + + + "used_skills": msg.used_skills or [], + "used_mcp": msg.used_mcp or [], + + + + "created_at": str(msg.created_at) if msg.created_at else None, + + + + }) + + + + + + + + return {"code": 0, "data": result, "message": "ok"} + +@router.post("/conversations/{conv_id}/messages") + + + +async def send_message( + + + + conv_id: str, + + + + req: SendMessageRequest, + + + + db: Session = Depends(get_db), + + + + current_user: User = Depends(get_current_user), + + + +): + 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) + + +@router.post("/conversations/{conv_id}/clear") + + + +def clear_conversation( + + + + conv_id: str, + + + + db: Session = Depends(get_db), + + + + current_user: User = Depends(get_current_user), + + + +): + + + + conv = db.query(Conversation).filter( + + + + and_(Conversation.id == conv_id, Conversation.user_id == current_user.id) + + + + ).first() + + + + if not conv: + + + + raise HTTPException(status_code=404, detail="对话不存在") + + + + + + + + db.query(Message).filter( + + + + Message.conversation_id == conv_id + + + + ).delete() + + + + + + + + conv.knowledge_graph_count = 0 + + + + conv.updated_at = cst_now() + + + + db.commit() + + + + + + + + return {"code": 0, "message": "对话上下文已清空"} + +@router.post("/conversations/{conv_id}/to-knowledge-graph") + + + +def convert_to_knowledge_graph( + + + + conv_id: str, + + + + db: Session = Depends(get_db), + + + + current_user: User = Depends(get_current_user), + + + +): + + + + """将对话内容转换为知识图谱（幂等：同一对话仅生成一次知识库）""" + + + + conv = db.query(Conversation).filter( + + + + and_(Conversation.id == conv_id, Conversation.user_id == current_user.id) + + + + ).first() + + + + if not conv: + + + + raise HTTPException(status_code=404, detail="对话不存在") + + + + + + + + messages = ( + + + + db.query(Message) + + + + .filter(Message.conversation_id == conv_id) + + + + .order_by(asc(Message.created_at), asc(Message.id)) + + + + .all() + + + + ) + + + + + + + + if not messages: + + + + raise HTTPException(status_code=400, detail="对话中没有消息, 无法生成知识图谱") + + + + + + + + # 幂等：若该对话已生成过知识图谱且知识库仍存在，则直接刷新返回 + + + + if conv.knowledge_graph_base_id: + + + + existing_kb = db.query(KnowledgeBase).filter( + + + + KnowledgeBase.id == conv.knowledge_graph_base_id + + + + ).first() + + + + if existing_kb: + + + + graph_data = _build_knowledge_graph(db, existing_kb) + + + + conv.knowledge_graph_count = 1 + + + + conv.updated_at = cst_now() + + + + db.commit() + + + + return { + + + + "code": 0, + + + + "data": { + + + + "knowledge_base_id": existing_kb.id, + + + + "knowledge_base_name": existing_kb.name, + + + + "graph": graph_data, + + + + "knowledge_graph_count": 1, + + + + "already_exists": True, + + + + }, + + + + "message": "知识图谱已存在, 已刷新展示", + + + + } + + + + + + + + kb_name = f"对话图谱-{conv.title}" + + + + kb = KnowledgeBase( + + + + id=gen_uuid(), + + + + user_id=current_user.id, + + + + name=kb_name, + + + + description=f"从对话「{conv.title}」自动生成的知识图谱", + + + + directory_path="/", + + + + document_count=0, + + + + created_at=cst_now(), + + + + updated_at=cst_now(), + + + + ) + + + + db.add(kb) + + + + db.flush() + + + + + + + + doc_count = 0 + + + + for idx, msg in enumerate(messages): + + + + role_label = "用户" if msg.role == "user" else "AI" + + + + preview = re.sub(r'[\\n\\r\\t]+', ' ', msg.content[:40]).strip() + + + + filename = f"msg_{idx+1:03d}_{role_label}_{preview}.txt" + + + + + + + + doc = KnowledgeDocument( + + + + id=gen_uuid(), + + + + knowledge_base_id=kb.id, + + + + filename=filename, + + + + file_type="txt", + + + + file_size=len(msg.content.encode('utf-8')), + + + + content=msg.content, + + + + status="completed", + + + + chunk_count=max(1, len(msg.content) // 500 + 1), + + + + created_at=cst_now(), + + + + updated_at=cst_now(), + + + + ) + + + + db.add(doc) + + + + doc_count += 1 + + + + + + + + kb.document_count = doc_count + + + + db.flush() + + + + + + + + # 构建知识图谱节点和边 + + + + graph_data = _build_knowledge_graph(db, kb) + + + + conv.knowledge_graph_base_id = kb.id + + + + conv.knowledge_graph_count = 1 + + + + conv.updated_at = cst_now() + + + + + + + + db.commit() + + + + + + + + LogService.log_operation( + + + + user_id=current_user.id, username=current_user.username, action="convert_to_graph", + + + + target="conversation", target_id=conv_id, + + + + detail=f"将对话 {conv.title} 转换为知识图谱 {kb.name}", + + + + ) + + + + + + + + return { + + + + "code": 0, + + + + "data": { + + + + "knowledge_base_id": kb.id, + + + + "knowledge_base_name": kb.name, + + + + "graph": graph_data, + + + + "knowledge_graph_count": 1, + + + + "already_exists": False, + + + + }, + + + + "message": f"知识图谱构建完成", + + + + } + +@router.get("/models") + + + +def get_models(db: Session = Depends(get_db)): + + + + models = db.query(AIModel).filter(AIModel.is_enabled == True).all() + + + + result = [] + + + + for m in models: + + + + provider = db.query(ModelProvider).filter(ModelProvider.id == m.provider_id).first() + + + + result.append({ + + + + "id": m.id, + + + + "name": m.name, + + + + "display_name": getattr(m, "display_name", None) or m.name, + + + + "provider_id": m.provider_id, + + + + "provider_name": provider.name if provider else "", + + + + "capabilities": m.capabilities or [], + + + + "status": m.status, + + + + "price_per_million_tokens": float(m.price_per_million_tokens or 0), + + + + "is_enabled": m.is_enabled, + + + + "is_default": bool(m.is_default) if hasattr(m, 'is_default') else False, + + + + "total_calls": m.total_calls or 0, + + + + "total_tokens": m.total_tokens or 0, + + + + "created_at": str(m.created_at) if m.created_at else None, + + + + }) + + + + result.sort(key=lambda x: (0 if x["is_default"] else 1, x["name"])) + + + + return {"code": 0, "data": result, "message": "ok"} diff --git a/backend/python/app/config.py b/backend/python/app/config.py index 50ffbb4..270e3a9 100644 --- a/backend/python/app/config.py +++ b/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,10 +81,14 @@ SLOW_REQUEST_THRESHOLD = float(os.getenv("SLOW_REQUEST_THRESHOLD", "3.0")) def setup_file_logging(): """配置应用日志按天轮转输出到文件""" - log_format = logging.Formatter( - "%(asctime)s [%(levelname)s] %(name)s - %(message)s", - datefmt="%Y-%m-%d %H:%M:%S", - ) + 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 [tid:%(trace_id)s] - %(message)s", + datefmt="%Y-%m-%d %H:%M:%S", + ) # 应用日志 - 按天轮转，保留30天 app_handler = TimedRotatingFileHandler( @@ -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 diff --git a/backend/python/app/core/__init__.py b/backend/python/app/core/__init__.py new file mode 100644 index 0000000..bfe8fd5 --- /dev/null +++ b/backend/python/app/core/__init__.py @@ -0,0 +1 @@ +# app/core package - 统一可观测基座 (Phase 2.1) diff --git a/backend/python/app/core/observability.py b/backend/python/app/core/observability.py new file mode 100644 index 0000000..1bce417 --- /dev/null +++ b/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", +] diff --git a/backend/python/main.py b/backend/python/main.py index 69aa41d..bd9905e 100644 --- a/backend/python/main.py +++ b/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(