From e8a1791c9f21601c8079184c8f2fdbd4b7fa2886 Mon Sep 17 00:00:00 2001 From: QihuAI Design Date: Fri, 24 Jul 2026 12:18:52 +0800 Subject: [PATCH] chore(align): sync Git with running container as authoritative source MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Standards alignment (STANDARDS_AUDIT_2026-07-24): Phase 1 标准化核心 + Phase 2 质量层 landed in the running container and captured in Git so the three sources (container := image := Git) agree. Backend (Phase 1 标准化核心): - mcp_client.py: replaced hand-rolled JSON-RPC with official `mcp` SDK (ClientSession, streamable_http/sse transports, protocol auto-negotiation) - core/otel.py + core/observability.py: standard OpenTelemetry SDK bridge (OTLP gRPC), wired to Grafana Alloy (qihu-aiflow-alloy:4317) which forwards metrics to Prometheus; emits aiflow_span_duration_ms / aiflow_span_count - deployments: Grafana Alloy container + Prometheus remote-write receiver Backend (Phase 2 质量层): - core/rag_pipeline.py: standardized RAG (recursive overlap chunking + rerank hook) reusing pgvector_search - core/llm_client.py: SDK-ified LLM + embeddings (aembed/embed) collapsing hand-rolled httpx calls; trust_env=False to bypass container proxy Migrations: - 018_normalize_marketapp_scene.sql: fixed capability allowlist (prevents capability wipe on restart) Alignment performed 2026-07-24 12:18:51. No deletions; secrets/venv/node_modules/ caches were not touched (tracked-source-only copy). --- backend/python/app/api/chat_service.py | 8581 +++++++++++----------- backend/python/app/api/im_llm.py | 565 +- backend/python/app/core/llm_client.py | 181 + backend/python/app/core/observability.py | 39 +- backend/python/app/core/otel.py | 131 + backend/python/app/core/rag_pipeline.py | 97 + backend/python/app/mcp_client.py | 185 +- backend/python/app/pgvector_search.py | 31 +- 8 files changed, 5031 insertions(+), 4779 deletions(-) create mode 100644 backend/python/app/core/llm_client.py create mode 100644 backend/python/app/core/otel.py create mode 100644 backend/python/app/core/rag_pipeline.py diff --git a/backend/python/app/api/chat_service.py b/backend/python/app/api/chat_service.py index 619e1cf..47c8678 100644 --- a/backend/python/app/api/chat_service.py +++ b/backend/python/app/api/chat_service.py @@ -1,4363 +1,4218 @@ -# Auto-extracted from chat.py during Phase 1 refactor (Strangler Fig). - -# Source of truth: scripts/phase1_extract.py. Do not hand-merge back. - - - -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 app.core.observability import span, get_trace_id - -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_external_agent) - -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) - -from app.api.chat_utils import (SendMessageRequest, _proxy_for_url, cst_now, logger) - - - -async def _send_message_core( - - - - - - - - conv_id: str, - - - - - - - - req: SendMessageRequest, - - - - - - - - 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="对话不存在") - - - - - - - - - - - - - - - - # Stage B.2: 多轮上下文保持 — 如果当前请求没有skill/agent/kb, - - - - - - - - # 从对话历史中继承上一次的选择 - - - - - - - - effective_skill_ids = req.skill_ids or [] - - - - - - - - effective_agent_ids = req.agent_ids or [] - - - - - - - - effective_kb_ids = req.knowledge_base_ids or [] - - - - - - - - effective_mcp_ids = req.mcp_ids or [] - - - - - - - - effective_use_tools = req.use_tools or False - - - - - - - - - - - - - - - - if not effective_skill_ids and not effective_agent_ids and not effective_kb_ids and not effective_mcp_ids: - - - - - - - - # 从对话上下文继承 - - - - - - - - prev_context = _get_conversation_context(db, conv_id) - - - - - - - - if prev_context: - - - - - - - - effective_skill_ids = prev_context.get("skill_ids", []) - - - - - - - - effective_agent_ids = prev_context.get("agent_ids", []) - - - - - - - - effective_kb_ids = prev_context.get("knowledge_base_ids", []) - - - - - - - - effective_mcp_ids = prev_context.get("mcp_ids", []) - - - - - - - - if effective_skill_ids or effective_agent_ids or effective_mcp_ids: - - - - - - - - effective_use_tools = True - - - - - - - - - - - - - - - - # 附件内容拼接到用户消息中(文本类文件附带内容, 供模型参考) - - - - - - - - attachment_block = "" - - - - - - - - if req.attachments: - - - - - - - - parts = [] - - - - - - - - for att in req.attachments: - - - - - - - - name = att.get("name", "附件") if isinstance(att, dict) else str(att) - - - - - - - - ftype = att.get("file_type", "") if isinstance(att, dict) else "" - - - - - - - - content = att.get("content", "") if isinstance(att, dict) else "" - - - - - - - - if content: - - - - - - - - parts.append(f"【附件: {name}】\n{content}") - - - - - - - - else: - - - - - - - - parts.append(f"【附件(二进制, 名称: {name}, 类型: {ftype})】") - - - - - - - - if parts: - - - - - - - - attachment_block = "\n\n" + "\n\n".join(parts) - - - - - - - - - - - - - - - - # 保存用户消息(含skill/agent/kb/mcp选择用于上下文保持) - - - - - - - - user_msg = Message( - - - - - - - - id=gen_uuid(), - - - - - - - - conversation_id=conv_id, - - - - - - - - role="user", - - - - - - - - content=req.content + attachment_block, - - - - - - - - token_used=len(req.content) // 2, - - - - - - - - model_id=req.model_id or conv.model_id, - - - - - - - - skill_ids=req.skill_ids if req.skill_ids else None, - - - - - - - - agent_ids=req.agent_ids if req.agent_ids else None, - - - - - - - - knowledge_ids=req.knowledge_base_ids if req.knowledge_base_ids else None, - - - - - - - - mcp_ids=req.mcp_ids if req.mcp_ids else None, - - - - - - - - attachments=req.attachments if req.attachments else None, - - - - - - - - created_at=cst_now(), - - - - - - - - ) - - - - - - - - db.add(user_msg) - - - - - - - - - - - - - - - - # ===== 模糊创作意图拦截 ===== - - - - - - - - # 用户使用创作类动词(制作/创作/做/生成…)但未明确产出形式时，先与用户确认方向， - - - - - - - - # 避免直接调用大模型被内容安全策略拒答(如返回 "high risk")。 - - - - - - - - clarify = detect_ambiguous_creative_intent(req.content) - - - - - - - - if clarify: - - - - - - - - clarify_content = "<>" + json.dumps( - - - - - - - - {"prompt": req.content, "options": clarify["options"]}, ensure_ascii=False - - - - - - - - ) + "<>" - - - - - - - - ai_msg = Message( - - - - - - - - id=gen_uuid(), - - - - - - - - conversation_id=conv_id, - - - - - - - - role="assistant", - - - - - - - - content=clarify_content, - - - - - - - - token_used=0, - - - - - - - - model_id=req.model_id or conv.model_id, - - - - - - - - created_at=cst_now(), - - - - - - - - ) - - - - - - - - db.add(ai_msg) - - - - - - - - db.commit() - - - - - - - - db.refresh(user_msg) - - - - - - - - db.refresh(ai_msg) - - - - - - - - return { - - - - - - - - "code": 0, - - - - - - - - "data": { - - - - - - - - "user_message": { - - - - - - - - "id": user_msg.id, "role": user_msg.role, "content": user_msg.content, - - - - - - - - "token_used": user_msg.token_used, "created_at": str(user_msg.created_at), - - - - - - - - }, - - - - - - - - "assistant_message": { - - - - - - - - "id": ai_msg.id, "role": ai_msg.role, "content": clarify_content, - - - - - - - - "created_at": str(ai_msg.created_at), "clarify": clarify, - - - - - - - - }, - - - - - - - - }, - - - - - - - - "message": "请选择创作方向", - - - - - - - - } - - - - - - - - - - - - - - - - # ===== 核心聊天引擎 ===== - - - - - - - - user_id = current_user.id - - - - - - - - model_id = req.model_id or conv.model_id - - - - - - - - prompt_tokens = 0 - - - - - - - - completion_tokens = 0 - - - - - - - - total_tokens = 0 - - - - - - - - ai_content = "" - - - - - - - - - - - - - - - - ai_model = None - - - - - - - - skills = [] # 初始化避免UnboundLocalError - process_steps = [] # 过程步骤(思考/技能/MCP执行)，用于前端收拢式展示 - - - - - - - - - agents = [] # 初始化避免UnboundLocalError - - - - - - - - media_generating = False # 提前初始化：外部Agent分支与末尾均会引用 - - - - - - - - external_handled = False # 是否已由外部Agent(Dify/n8n/Coze)远程处理 - - - - - - - - - - - - - - - - # ===== 外部 Agent 远程调用分支 (Dify / n8n / Coze) ===== - - - - - - - - # 若本轮对话关联了带有效 agent_endpoint 的外部 Agent，则直接远程调用其平台， - - - - - - - - # 跳过本地 LLM「扮演」逻辑。effective_agent_ids 已在前面从请求/上下文继承。 - - - - - - - - external_agents = _resolve_market_agents(db, effective_agent_ids, user_id) - - - - - - - - external_agent = next((a for a in external_agents if a.get("agent_endpoint")), None) - - - - # ===== 3.3 多 Agent 编排(图引擎)快路 ===== - - _orchestrator = next((a for a in external_agents if a.get("orchestration")), None) - - if _orchestrator: - - try: - - from app.core.orchestrator import run_graph - - _answer, _trace = await run_graph( - - graph=_orchestrator["orchestration"], - - query=req.content + attachment_block, - - db=db, - - user_id=user_id, - - conv_id=conv_id, - - trace_id=get_trace_id(), - - ) - - ai_content = _answer or "编排未产生内容。" - - logger.info( - - f"[orchestrator] 图执行完成 trace_id={get_trace_id()} nodes={len(_trace.get('nodes', []))}" - - ) - - except Exception as e: - - logger.warning(f"[orchestrator] 图执行失败: {e}") - - ai_content = f"⚠️ 多 Agent 编排执行失败：{str(e)[:200]}" - - external_handled = True - - external_agent = None # 阻止下方外部 Agent 分支重复执行 - - - - - - - - if external_agent: - - - - - - - - agents = external_agents - - - - - - - - # 组装最近历史(供 n8n 等无平台会话态的服务维持多轮上下文) - - - - - - - - _history = _build_chat_history(db, conv_id, max_tokens=4000, exclude_latest_user=True) - - - - - - - - try: - - - - - - - - answer, new_ext_id = await _call_external_agent( - - - - - - - - db, external_agent, - - - - - - - - query=req.content + attachment_block, - - - - - - - - user_id=user_id, - - - - - - - - ext_conversation_id=conv.ext_agent_conversation_id, - - - - - - - - history=_history, - - - - - - - - ) - - - - - - - - ai_content = answer or "外部 Agent 未返回内容。" - - - - - - - - if new_ext_id: - - - - - - - - conv.ext_agent_conversation_id = new_ext_id - - - - - - - - except Exception as e: - - - - - - - - logger.warning( - - - - - - - - f"[external_agent] 远程调用失败 platform={external_agent.get('platform')} " - - - - - - - - f"endpoint={external_agent.get('agent_endpoint')}: {e}" - - - - - - - - ) - - - - - - - - ai_content = f"⚠️ 外部 Agent 调用失败：{str(e)[:200]}" - - - - - - - - external_handled = True - - - - - - - - - - - - - - - - # 解析模型：支持 Auto 模式（model_id='auto' 或未指定时，按能力自动匹配平台默认模型） - - - - - - - - if not external_handled: - - - - - - - - if model_id and model_id != "auto": - - - - - - - - ai_model = db.query(AIModel).filter(AIModel.id == model_id).first() - - - - - - - - if not ai_model: - - - - - - - - # Auto / 未指定 → 按用户输入意图匹配对应能力的默认模型，否则回退文本对话 - - - - - - - - intent = detect_media_intent(req.content) - - - - - - - - resolved = resolve_auto_model(intent, db) if intent else None - - - - - - - - ai_model = resolved or resolve_auto_model("text", db) - - - - - - - - - - - - - - - - - - - - # ===== Model Failover: Auto 模式下自动切换备选模型 ===== - _try_models = [ai_model] if (ai_model and ai_model.is_enabled) else [] - if not model_id or model_id == "auto": - # Auto 模式：附加其他已启用的文本模型作为备选 - _fallbacks = db.query(AIModel).filter( - AIModel.is_enabled == True, - AIModel.id != (ai_model.id if ai_model else "") - ).order_by(AIModel.is_default.desc()).limit(5).all() - _try_models.extend([m for m in _fallbacks if m not in _try_models]) - _retry_log = [] - _all_failed = False - - for _try_model in _try_models: - ai_model = _try_model - if ai_model: - - - - - - - - # ====== 图像/视频生成模型分支 ====== - - - - - - - - caps = [c.lower() for c in (ai_model.capabilities or [])] if ai_model else [] - - - - - - - - media_generating = False - - - - - - - - if ai_model and ("image" in caps or "video" in caps): - - - - - - - - # 异步生成：立即返回占位内容，后台任务轮询 AGNES 并回写， - - - - - - - - # 避免最长 5 分钟的视频轮询阻塞 HTTP 请求导致客户端超时断开。 - - - - - - - - media_generating = True - - - - - - - - if "video" in caps: - - - - - - - - ai_content = "⏳ 视频生成中…（通常 1~3 分钟，完成后自动刷新）" - - - - - - - - else: - - - - - - - - ai_content = "⏳ 图像生成中…（完成后自动刷新）" - - - - - - - - elif ai_model and ai_model.endpoint: - - - - - - - - provider = db.query(ModelProvider).filter(ModelProvider.id == ai_model.provider_id).first() - - - - - - - - api_key = ai_model.api_key or (provider.api_key if provider else None) - - - - - - - - api_url = ai_model.endpoint - - - - - - - - - - - - - - - - if not api_url and provider and provider.api_base_url: - - - - - - - - api_url = f"{provider.api_base_url.rstrip('/')}/v1/chat/completions" - - - - - - - - - - - - - - - - if api_key and api_url: - - - - - - - - try: - - - - - - - - # Stage A+B: 解析Skills和Agents - - - - - - - - skills = _resolve_market_skills(db, effective_skill_ids, user_id) - - - - - - - - agents = _resolve_market_agents(db, effective_agent_ids, user_id) - - - - - - - - mcps = _resolve_active_mcp_servers(db, effective_mcp_ids) - - - - - - - - # MCP 工具调用链路：拉取各 MCP 工具并转为 function definitions - - - - mcp_servers_conn = mcps - - - - mcp_function_definitions = [] - - - - mcp_tool_map = {} - - - - mcp_idx = 0 - - - - for _srv in mcp_servers_conn: - - - - 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: - - - - logger.warning(f"MCP 拉取工具失败 [{_srv.get('name')}]: {_e}") - - - - _tools = [] - - - - if _tools: - - - - try: - - - - _upd = db.query(MCPServer).filter(MCPServer.id == _srv["id"]).first() - - - - if _upd: - - - - _upd.tools_count = len(_tools) - - - - db.commit() - - - - except Exception: - - - - pass - - - - for _t in _tools: - - - - _fname = f"mcp_tool_{mcp_idx}" - - - - mcp_tool_map[_fname] = { - - - - "server_id": _srv["id"], - - - - "name": _t.get("name"), - - - - "server_name": _srv["name"], - - - - } - - - - _schema = _t.get("inputSchema") or {"type": "object", "properties": {}} - - - - if not isinstance(_schema, dict) or _schema.get("type") != "object": - - - - _schema = {"type": "object", "properties": {}} - - - - mcp_function_definitions.append({ - - - - "name": _fname, - - - - "description": f"[MCP·{_srv['name']}] {_t.get('description', '')}", - - - - "parameters": _schema, - - - - }) - - - - mcp_idx += 1 - - - - if mcp_function_definitions: - - - - effective_use_tools = True - - - - - - - - - - - - - - - - # Issue 8: 默认技能自动发现（仅当用户未显式选择技能时启用） - - - - - - - - # 每轮先寻找本账号已安装且平台已部署的技能，多个适用时由模型自行判断选 1 个最优。 - - - - - - - - used_auto_skills = False - - - - - - - - if not effective_skill_ids: - - - - - - - - try: - - - - - - - - discovered = _discover_installed_skills(db, user_id) - - - - - - - - except Exception as e: - - - - - - - - logger.warning(f"自动发现已安装技能失败: {e}") - - - - - - - - discovered = [] - - - - - - - - if discovered: - - - - - - - - skills = discovered - - - - - - - - used_auto_skills = True - - - - - - - - effective_use_tools = True - - - - - - - - - - - - - - - - # Stage B.1: Agent可能自带skill_ids和kb_ids - - - - - - - - agent_extra_skill_ids = [] - - - - - - - - agent_extra_kb_ids = [] - - - - - - - - agent_model_override = None - - - - - - - - for agent in agents: - - - - - - - - if agent.get("skill_ids"): - - - - - - - - agent_extra_skill_ids.extend(agent["skill_ids"]) - - - - - - - - if agent.get("knowledge_base_ids"): - - - - - - - - agent_extra_kb_ids.extend(agent["knowledge_base_ids"]) - - - - - - - - if agent.get("model_id"): - - - - - - - - agent_model_override = agent["model_id"] - - - - - - - - - - - - - - - - # 合并Agent自带的skills和KBs - - - - - - - - all_skill_ids = list(set(effective_skill_ids + agent_extra_skill_ids)) - - - - - - - - all_kb_ids = list(set(effective_kb_ids + agent_extra_kb_ids)) - - - - - - - - - - - - - - - - if agent_extra_skill_ids or agent_extra_kb_ids: - - - - - - - - # 重新解析合并后的skills - - - - - - - - skills = _resolve_market_skills(db, all_skill_ids, user_id) - - - - - - - - effective_use_tools = True - - - - - - - - - - - - - - - - # Stage C: 增强版KB检索 - - - - - - - - kb_context = await _retrieve_kb_context(db, all_kb_ids, req.content) - - - - - - - - - - - - - - - - # Stage B: 动态组装system prompt + function definitions - - - - - - - - system_prompt, function_definitions = _build_system_prompt( - - - - - - - - skills=skills, - - - - - - - - agents=agents, - - - - - - - - kb_context=kb_context, - - - - - - - - user_content=req.content, - - - - - - - - mcp=mcps, - - - - mcp_functions=mcp_function_definitions, - - - - auto_discovered=used_auto_skills, - - all_mcp=_list_enabled_mcp_servers(db), - - db=db, - - ) - - - - - - - - - - - - - - - - # 构建消息列表 - - - - - - - - # 实际调用追踪（Issue：反馈真实使用的技能/MCP，而非注入的全部） - - called_skill_names = [] - - called_mcp_names = [] - - - - messages = [{"role": "system", "content": system_prompt}] - - - - - - - - - - - - - - - - # 分层上下文引擎：预算感知历史 + 压缩摘要(若有) - - messages.extend(_build_chat_history(db, conv_id, max_tokens=6000, exclude_latest_user=True)) - - - - - - - - - - - - - - - - # Issue 1: 必须把附件内容(attachment_block)一并发给模型， - - - - - - - - # 否则用户在对话窗上传文件后模型“看不到”文件内容。 - - - - - - - - # Issue 2 修复：用户输入含 URL 且要求读取/加入/安装时，服务端抓取真实页面内容注入上下文 - try: - from app.api.url_reader import fetch_url_context - _url_ctx = await fetch_url_context(req.content) - except Exception as _e: - import logging as _logging - _logging.getLogger(__name__).warning("URL 上下文抓取失败(已跳过): %s", _e) - _url_ctx = None - if _url_ctx and _url_ctx.get("text"): - messages.append({ - "role": "system", - "content": ( - "【用户随消息提供的参考链接的真实内容】\n" - f"链接地址: {_url_ctx.get('url')}\n" - "下面是该链接页面的真实内容，请严格基于它来回答用户的需求" - "（例如按其『Quick Start / 加入步骤』逐步引导用户完成安装或加入），" - "不要凭空编造步骤：\n\n" + _url_ctx["text"] - ), - }) - process_steps.append({ - "type": "skill", - "name": "网页/仓库读取", - "result": f"已读取 {_url_ctx.get('url')}（{len(_url_ctx['text'])} 字符）", - }) - messages.append({"role": "user", "content": req.content + attachment_block}) - - - - - - - - - - - - - - - - # Stage B.3: Function Calling循环(最多5轮) - - - - - - - - max_tool_rounds = 5 - - - - - - - - # ===== 3.1 tool_policy 沙箱：聚合策略 + 初始化预算计数器 ===== - - _agent_tool_policy = _resolve_agent_tool_policy(agents) - - _tool_call_count = 0 - - for round_num in range(max_tool_rounds): - - _round_tool_count = 0 - - - - - - - - request_body = { - - - - - - - - "model": ai_model.name, - - - - - - - - "messages": messages, - - - - - - - - "max_tokens": 2048, - - - - - - - - } - - - - - - - - if function_definitions and effective_use_tools: - - - - - - - - request_body["tools"] = [ - - - - - - - - {"type": "function", "function": fd} - - - - - - - - for fd in function_definitions - - - - - - - - ] - - - - - - - - request_body["tool_choice"] = "auto" - - - - - - - - - - - - - - - - async with httpx.AsyncClient( - - - - - - - - timeout=ai_model.timeout or 120, - - - - - - - - proxy=_proxy_for_url(api_url), - - - - - - - - trust_env=False, - - - - - - - - ) as client: - - - - - - - - async with span("llm.completion", model=ai_model.id if ai_model else None): - - resp = await client.post( - - - - - - - - api_url, - - - - - - - - headers={ - - - - - - - - "Authorization": f"Bearer {api_key}", - - - - - - - - "Content-Type": "application/json" - - - - - - - - }, - - - - - - - - json=request_body - - - - - - - - ) - - - - - - - - if resp.status_code != 200: - - - - - - - - if round_num == 0: - - - - - - - - ai_content = f"AI模型调用失败 (HTTP {resp.status_code}), 请检查模型配置。" - - - - - - - - break - - - - - - - - - - - - - - - - data = resp.json() - - - - - - - - choice = data.get("choices", [{}])[0] - - - - - - - - message = choice.get("message", {}) - - - - - - - - - - - - - - - - # 记录token使用量 - - - - - - - - usage = data.get("usage", {}) - - - - - - - - prompt_tokens += usage.get("prompt_tokens", 0) - - - - - - - - completion_tokens += usage.get("completion_tokens", 0) - - - - - - - - total_tokens += usage.get("total_tokens", 0) - - - - - - - - - - - - - - - - # 检查是否有tool_calls - - - - - - - - tool_calls = message.get("tool_calls", []) - # 过程步骤：捕获模型思考/推理(产生工具调用前的推理) - _think = (message.get("content") or "").strip() - if _think and tool_calls: - process_steps.append({ - "type": "thinking", - "text": _think, - }) - - - - - - - - - if tool_calls and effective_use_tools: - - - - - - - - # 将assistant消息(含tool_calls)加入上下文 - - - - - - - - messages.append({ - - - - - - - - "role": "assistant", - - - - - - - - "content": message.get("content") or "", - - - - - - - - "tool_calls": tool_calls, - - - - - - - - }) - - - - - - - - - - - - - - - - # Stage A.2: 真实Skill执行 - - - - - - - - for tc in tool_calls: - - - - - - - - func_name = tc.get("function", {}).get("name", "") - - # ===== 3.1 tool_policy 沙箱：逐工具校验（白名单 + 预算 + 每轮上限） ===== - - _mcp_info = mcp_tool_map.get(func_name) if func_name.startswith("mcp_tool_") else None - - _policy_check = _check_tool_policy(func_name, _mcp_info, skills, _agent_tool_policy, _tool_call_count, _round_tool_count) - - if not _policy_check["allowed"]: - - _reason = _policy_check["reason"] - - logger.warning( - - f"[tool_policy] blocked tool call: agent={_agent_tool_policy.get('agent_id') if _agent_tool_policy else None} " - - f"tool={func_name} reason={_reason} trace={get_trace_id()}" - - ) - - tool_result = f"[工具调用已被策略拦截] {_policy_check.get('message', _reason)}" - - messages.append({ - - "role": "tool", - - "tool_call_id": tc.get("id", f"call_{round_num}"), - - "content": tool_result, - - }) - - _tool_call_count += 1 - - _round_tool_count += 1 - - continue - - _tool_call_count += 1 - - _round_tool_count += 1 - - - - - - - - func_args_str = tc.get("function", {}).get("arguments", "{}") - - - - - - - - try: - - - - - - - - func_args = json.loads(func_args_str) - - - - - - - - except Exception: - - - - - - - - func_args = {"input": func_args_str} - - - - - - - - - - - - - - - - if func_name.startswith("mcp_tool_"): - - - - # MCP 工具调用：代理到对应 MCP 服务执行 tools/call - - - - _info = mcp_tool_map.get(func_name) - - - - tool_result = "[MCP 工具未找到或映射缺失]" - - - - if _info: - - - - _srv = next( - - - - (s for s in mcp_servers_conn if s["id"] == _info["server_id"]), - - - - None, - - - - ) - - - - if _srv: - - - - async with span(f"mcp.call:{_info.get('name') if isinstance(_info, dict) else None}"): - - tool_result = await call_mcp_tool( - - - - _srv, _info["name"], func_args - - - - ) - - - - called_mcp_names.append(_srv.get("name", "")) - process_steps.append({ - "type": "mcp", - "name": _info.get("name") if isinstance(_info, dict) else func_name, - "server": _srv.get("name", ""), - "args": func_args, - "result": (tool_result or "")[:800], - }) - - - - - else: - - - - async with span(f"skill.execute:{func_name}"): - - tool_result = _execute_skill_tool( - - - - db, func_name, func_args, skills, req.content - - - - ) - - - - _sk = next((s for s in skills if s.get("func_name") == func_name), None) - - - - if _sk: - - - - called_skill_names.append(_sk.get("name", "")) - process_steps.append({ - "type": "skill", - "name": (_sk.get("name") if _sk else func_name), - "args": func_args, - "result": (tool_result or "")[:800], - }) - - - - - - - - - messages.append({ - - - - - - - - "role": "tool", - - - - - - - - "tool_call_id": tc.get("id", f"call_{round_num}"), - - - - - - - - "content": tool_result, - - - - - - - - }) - - - - - - - - - - - - - - - - # 继续下一轮让模型处理tool结果 - - - - - - - - continue - - - - - - - - else: - - - - - - - - # 普通回复 - - - - - - - - ai_content = message.get("content", "") - - - - - - - - if not ai_content and round_num > 0: - - - - - - - - ai_content = "AI模型已处理工具调用结果, 但返回了空响应。" - - - - - - - - break - - - - - - - - - - - - - - - - if not ai_content: - - - - - - - - ai_content = "AI模型返回了空响应, 请稍后重试。" - - - - - - - - - - - - - - - - except Exception as e: - - - - - - - - logger.warning(f"AI API调用异常: {e}") - - - - - - - - ai_content = "AI服务暂时不可用, 请稍后重试或检查模型API配置。" - - - - - - - - else: - - - - - - - - ai_content = "该模型缺少API密钥或端点配置, 请在后台管理-大模型库中配置API Key和Endpoint。" - - - - - - - - else: - - - - - - - - ai_content = "请选择一个已配置的AI模型进行对话。" - - - - - - - - elif not external_handled: - - - - - - - - ai_content = "请先在后台管理-大模型库配置默认对话模型（或聊天中选择模型），Auto 模式下将自动使用平台默认模型。" - - - - - # ===== Post-check: 失败则切换备选，成功则跳出 ===== - _is_error = ( - ai_content and ( - "AI模型调用失败" in str(ai_content) - or "AI服务暂时不可用" in str(ai_content) - or "AI模型返回了空响应" in str(ai_content) - or "该模型缺少API密钥" in str(ai_content) - ) - ) - if _is_error: - _retry_log.append(f"{ai_model.name}: {str(ai_content)[:60]}") - logger.warning( - f"[Failover] 模型 {ai_model.name} 失败，切换至备选..." - ) - continue # try next model - - # Success — exit failover loop - break - else: - # Loop completed without break = all models failed - _all_failed = True - - # ===== All models failed? ===== - if _all_failed: - _names = ", ".join([m.name for m in _try_models]) - _details = "; ".join(_retry_log[-3:]) if _retry_log else "未知错误" - ai_content = ( - f"已尝试 {len(_try_models)} 个模型（{_names}）均调用失败。\n" - f"最后错误: {_details}\n" - f"请稍后重试或检查模型配置。" - ) - logger.error(f"[Failover] 全部 {len(_try_models)} 个模型均失败: {_details}") - - - - - - - - - - - - - - - - - - - - - ai_token_count = len(ai_content) // 2 - - - - - - - - ai_msg = Message( - - - - - - - - id=gen_uuid(), - - - - - - - - conversation_id=conv_id, - - - - - - - - role="assistant", - - - - - - - - content=ai_content, - - - - - - - - token_used=ai_token_count, - - - - - - - - model_id=ai_model.id if ai_model else (req.model_id or conv.model_id), - - - - - - - - skill_ids=req.skill_ids if req.skill_ids else None, - - - - - - - - agent_ids=req.agent_ids if req.agent_ids else None, - - - - - - - - knowledge_ids=req.knowledge_base_ids if req.knowledge_base_ids else None, - - - - - - - - created_at=cst_now(), - - - - - - - - ) - - - - - - - - db.add(ai_msg) - - - - - - - - - - - - - - - - # 后台异步：将实质性文本回答入库到 Dify 知识库（量化后可供 RAG 复用） - - - - - - - - _ingest_text = (ai_content or "").strip() - - - - - - - - if len(_ingest_text) > 120 and not ( - - - - - - - - _ingest_text.startswith("[MEDIA") - - - - - - - - or _ingest_text.startswith("data:") - - - - - - - - or _ingest_text.startswith("http") - - - - - - - - or " 反馈配对的技能名 - - - - - - - - try: - - - - - - - - _sel = _resolve_market_skills(db, effective_skill_ids, user_id) - - - - - - - - used_skills = [s["name"] for s in _sel] - - - - - - - - except Exception: - - - - - - - - used_skills = [] - - - - - - - - elif called_skill_names: - - - - - - - - # 未显式选技能 -> 反馈 AI 本轮实际调用的技能（可能为空） - - - - - - - - used_skills = list(dict.fromkeys(called_skill_names)) - - - - - - - - if effective_mcp_ids: - - - - - - - - # 用户显式选择了 MCP -> 反馈配对的 MCP 名 - - - - - - - - try: - - - - - - - - _sel_mcp = _resolve_active_mcp_servers(db, effective_mcp_ids) - - - - - - - - used_mcp = [m["name"] for m in _sel_mcp] - - - - - - - - except Exception: - - - - - - - - used_mcp = [] - - - - - - - - elif called_mcp_names: - - - - - - - - used_mcp = list(dict.fromkeys(called_mcp_names)) - - - - - - - - ai_msg.used_skills = used_skills - - - - - - - - ai_msg.used_mcp = used_mcp - - - - - - - - - - - - - - - - # 更新对话统计 - - - - - - - - # 注意：knowledge_graph_count 仅应在用户实际点击「转知识图谱」时 +1， - # 不可在每次 AI 回复时无条件自增，否则未操作也会累积计数（见 #715）。 - - - - - - - - conv.updated_at = cst_now() - - - - - - - - - - - - - - - - # 记录Token使用（媒体生成走 _generate_media_task 异步补写真实消耗，此处跳过避免占位文本重复记账） - - - - - - - - if not media_generating: - - - - - - - - token_usage = TokenUsageLog( - - - - - - - - id=gen_uuid(), - - - - - - - - user_id=current_user.id, - - - - - - - - model_id=ai_model.id if ai_model else (req.model_id or conv.model_id), - - - - - - - - model_name=ai_model.name if ai_model else None, - - - - - - - - conversation_id=conv_id, - - trace_id=get_trace_id(), - - - - - - - - - - prompt_tokens=prompt_tokens or len(req.content) // 2, - - - - - - - - completion_tokens=completion_tokens or ai_token_count, - - - - - - - - total_tokens=total_tokens or (len(req.content) // 2 + ai_token_count), - - - - - - - - created_at=cst_now(), - - - - - - - - ) - - - - - - - - db.add(token_usage) - - - - - - - - - - - - - - - - # 进化轨迹记录 - - - - - - - - try: - - - - - - - - existing = db.query(EvolutionTrajectory).filter( - - - - - - - - EvolutionTrajectory.conversation_id == conv_id - - - - - - - - ).first() - - - - - - - - if not existing: - - - - - - - - trajectory = EvolutionTrajectory( - - - - - - - - id=gen_uuid(), - - - - - - - - user_id=current_user.id, - - - - - - - - conversation_id=conv_id, - - - - - - - - status="discovered", - - - - - - - - token_used=ai_token_count, - - - - - - - - created_at=cst_now(), - - - - - - - - ) - - - - - - - - db.add(trajectory) - - - - - - - - db.flush() - - - - - - - - - - - - - - - - try: - - - - - - - - from app.services.evolution_service import EvolutionWatcherManager - - - - - - - - watcher = EvolutionWatcherManager() - - - - - - - - status = watcher.get_user_status(current_user.id, lambda: SessionLocal()) - - - - - - - - if not status.get("scheduler_running"): - - - - - - - - watcher.start_user_evolution(current_user.id, lambda: SessionLocal()) - - - - - - - - logger.info(f"[Evolution] 自动启动用户 {current_user.id} 的进化Pipeline") - - - - - - - - except Exception as evolution_err: - - - - - - - - logger.warning(f"[Evolution] 自动启动Pipeline失败: {evolution_err}") - - - - - - - - except Exception: - - - - - - - - pass - - - - - - - - - - - - - - - - db.commit() - - - - - - - - # 后台异步压缩历史(分层记忆)，不阻塞本次响应 - - asyncio.create_task(_maybe_compress_history(conv_id)) - - - - db.refresh(user_msg) - - - - - - - - db.refresh(ai_msg) - - - - - - - - - - - - - - - - # 媒体生成改为后台异步：立即返回占位，后台任务完成后回写消息内容 - - - - - - - - if media_generating: - - - - - - - - try: - - - - - - - - asyncio.create_task( - - - - - - - - _generate_media_task(ai_msg.id, ai_model.id, req.content, current_user.id) - - - - - - - - ) - - - - - - - - except Exception as e: - - - - - - - - logger.warning(f"[_generate_media_task] 启动后台任务失败: {e}", exc_info=True) - # 启动失败兜底：不让占位消息永久停在「⏳ 生成中…」 - try: - ai_msg.content = "⚠️ 媒体生成启动失败，请稍后重试" - db.commit() - except Exception: - pass - - - - - - - - - - - - - - - - return { - - - - - - - - "code": 0, - - - - - - - - "data": { - - - - - - - - "user_message": { - - - - - - - - "id": user_msg.id, "role": user_msg.role, "content": user_msg.content, - - - - - - - - "token_used": user_msg.token_used, "created_at": str(user_msg.created_at), - - - - - - - - }, - - - - - - - - "assistant_message": { - - - - - - - - "id": ai_msg.id, "role": ai_msg.role, "content": ai_msg.content, - - - - - - - - "token_used": ai_msg.token_used, "created_at": str(ai_msg.created_at), - - - - - - - - "used_skills": used_skills, - - "used_mcp": used_mcp, - - - - - - - - "generating": media_generating, - "process_steps": process_steps, - - - - - - - - - }, - - - - - - - - }, - - - - - - - - "message": "发送成功", - - - - - - - - } - - - - - -# ====================================================================== - -# 3.1 tool_policy 沙箱辅助函数 - -# ====================================================================== - -def _resolve_agent_tool_policy(agents): - - """从已解析的 agents 聚合工具策略。多 agent 时取首个带有非空策略者。""" - - if not agents: - - return None - - for ag in agents: - - tp = ag.get("tool_policy") - - if tp and isinstance(tp, dict): - - tp = dict(tp) - - tp["agent_id"] = ag.get("id") - - return tp - - return None - - - - - -def _check_tool_policy(func_name, mcp_info, skills, policy, call_count, round_count): - - """校验单个工具调用是否被策略允许。 - - - - 返回 {"allowed": bool, "reason": str, "tool_id": str|None, "message": str} - - policy 为 None/空 -> 允许（向后兼容）。 - - """ - - if not policy: - - return {"allowed": True, "tool_id": None} - - allowed = policy.get("allowed_tools") or {} - - skill_ids = allowed.get("skill_ids") or [] - - mcp_ids = allowed.get("mcp_ids") or [] - - is_mcp = func_name.startswith("mcp_tool_") - - tool_id = None - - if is_mcp: - - if mcp_info and isinstance(mcp_info, dict): - - tool_id = mcp_info.get("server_id") - - if mcp_ids and tool_id not in mcp_ids: - - return {"allowed": False, "reason": "mcp_not_in_allowlist", "tool_id": tool_id, - - "message": "MCP 服务不在该智能体的允许列表中"} - - else: - - if skills: - - _s = next((s for s in skills if s.get("func_name") == func_name), None) - - if _s: - - tool_id = _s.get("id") - - if skill_ids and tool_id not in skill_ids: - - return {"allowed": False, "reason": "skill_not_in_allowlist", "tool_id": tool_id, - - "message": "技能不在该智能体的允许列表中"} - - budget = policy.get("max_calls_per_conversation") - - if budget is not None and call_count >= int(budget): - - return {"allowed": False, "reason": "budget_exceeded", "tool_id": tool_id, - - "message": policy.get("fallback_message") - - or "该智能体已达到工具调用上限，请开启新对话或调整智能体设置。"} - - per_round = policy.get("per_round_cap") - - if per_round is not None and round_count >= int(per_round): - - return {"allowed": False, "reason": "per_round_cap", "tool_id": tool_id, - - "message": "本轮工具调用已达上限"} - - return {"allowed": True, "tool_id": tool_id} - +# Auto-extracted from chat.py during Phase 1 refactor (Strangler Fig). + +# Source of truth: scripts/phase1_extract.py. Do not hand-merge back. + + + +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 app.core.observability import span, get_trace_id + +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_external_agent) + +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) + +from app.api.chat_utils import (SendMessageRequest, _proxy_for_url, cst_now, logger) +# 统一模型层(07-24 红线)：web chat 与 IM bot 共用标准 OpenAI SDK 客户端 +from app.core.llm_client import achat_complete + + + +async def _send_message_core( + + + + + + + + conv_id: str, + + + + + + + + req: SendMessageRequest, + + + + + + + + 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="对话不存在") + + + + + + + + + + + + + + + + # Stage B.2: 多轮上下文保持 — 如果当前请求没有skill/agent/kb, + + + + + + + + # 从对话历史中继承上一次的选择 + + + + + + + + effective_skill_ids = req.skill_ids or [] + + + + + + + + effective_agent_ids = req.agent_ids or [] + + + + + + + + effective_kb_ids = req.knowledge_base_ids or [] + + + + + + + + effective_mcp_ids = req.mcp_ids or [] + + + + + + + + effective_use_tools = req.use_tools or False + + + + + + + + + + + + + + + + if not effective_skill_ids and not effective_agent_ids and not effective_kb_ids and not effective_mcp_ids: + + + + + + + + # 从对话上下文继承 + + + + + + + + prev_context = _get_conversation_context(db, conv_id) + + + + + + + + if prev_context: + + + + + + + + effective_skill_ids = prev_context.get("skill_ids", []) + + + + + + + + effective_agent_ids = prev_context.get("agent_ids", []) + + + + + + + + effective_kb_ids = prev_context.get("knowledge_base_ids", []) + + + + + + + + effective_mcp_ids = prev_context.get("mcp_ids", []) + + + + + + + + if effective_skill_ids or effective_agent_ids or effective_mcp_ids: + + + + + + + + effective_use_tools = True + + + + + + + + + + + + + + + + # 附件内容拼接到用户消息中(文本类文件附带内容, 供模型参考) + + + + + + + + attachment_block = "" + + + + + + + + if req.attachments: + + + + + + + + parts = [] + + + + + + + + for att in req.attachments: + + + + + + + + name = att.get("name", "附件") if isinstance(att, dict) else str(att) + + + + + + + + ftype = att.get("file_type", "") if isinstance(att, dict) else "" + + + + + + + + content = att.get("content", "") if isinstance(att, dict) else "" + + + + + + + + if content: + + + + + + + + parts.append(f"【附件: {name}】\n{content}") + + + + + + + + else: + + + + + + + + parts.append(f"【附件(二进制, 名称: {name}, 类型: {ftype})】") + + + + + + + + if parts: + + + + + + + + attachment_block = "\n\n" + "\n\n".join(parts) + + + + + + + + + + + + + + + + # 保存用户消息(含skill/agent/kb/mcp选择用于上下文保持) + + + + + + + + user_msg = Message( + + + + + + + + id=gen_uuid(), + + + + + + + + conversation_id=conv_id, + + + + + + + + role="user", + + + + + + + + content=req.content + attachment_block, + + + + + + + + token_used=len(req.content) // 2, + + + + + + + + model_id=req.model_id or conv.model_id, + + + + + + + + skill_ids=req.skill_ids if req.skill_ids else None, + + + + + + + + agent_ids=req.agent_ids if req.agent_ids else None, + + + + + + + + knowledge_ids=req.knowledge_base_ids if req.knowledge_base_ids else None, + + + + + + + + mcp_ids=req.mcp_ids if req.mcp_ids else None, + + + + + + + + attachments=req.attachments if req.attachments else None, + + + + + + + + created_at=cst_now(), + + + + + + + + ) + + + + + + + + db.add(user_msg) + + + + + + + + + + + + + + + + # ===== 模糊创作意图拦截 ===== + + + + + + + + # 用户使用创作类动词(制作/创作/做/生成…)但未明确产出形式时，先与用户确认方向， + + + + + + + + # 避免直接调用大模型被内容安全策略拒答(如返回 "high risk")。 + + + + + + + + clarify = detect_ambiguous_creative_intent(req.content) + + + + + + + + if clarify: + + + + + + + + clarify_content = "<>" + json.dumps( + + + + + + + + {"prompt": req.content, "options": clarify["options"]}, ensure_ascii=False + + + + + + + + ) + "<>" + + + + + + + + ai_msg = Message( + + + + + + + + id=gen_uuid(), + + + + + + + + conversation_id=conv_id, + + + + + + + + role="assistant", + + + + + + + + content=clarify_content, + + + + + + + + token_used=0, + + + + + + + + model_id=req.model_id or conv.model_id, + + + + + + + + created_at=cst_now(), + + + + + + + + ) + + + + + + + + db.add(ai_msg) + + + + + + + + db.commit() + + + + + + + + db.refresh(user_msg) + + + + + + + + db.refresh(ai_msg) + + + + + + + + return { + + + + + + + + "code": 0, + + + + + + + + "data": { + + + + + + + + "user_message": { + + + + + + + + "id": user_msg.id, "role": user_msg.role, "content": user_msg.content, + + + + + + + + "token_used": user_msg.token_used, "created_at": str(user_msg.created_at), + + + + + + + + }, + + + + + + + + "assistant_message": { + + + + + + + + "id": ai_msg.id, "role": ai_msg.role, "content": clarify_content, + + + + + + + + "created_at": str(ai_msg.created_at), "clarify": clarify, + + + + + + + + }, + + + + + + + + }, + + + + + + + + "message": "请选择创作方向", + + + + + + + + } + + + + + + + + + + + + + + + + # ===== 核心聊天引擎 ===== + + + + + + + + user_id = current_user.id + + + + + + + + model_id = req.model_id or conv.model_id + + + + + + + + prompt_tokens = 0 + + + + + + + + completion_tokens = 0 + + + + + + + + total_tokens = 0 + + + + + + + + ai_content = "" + + + + + + + + + + + + + + + + ai_model = None + + + + + + + + skills = [] # 初始化避免UnboundLocalError + process_steps = [] # 过程步骤(思考/技能/MCP执行)，用于前端收拢式展示 + + + + + + + + + agents = [] # 初始化避免UnboundLocalError + + + + + + + + media_generating = False # 提前初始化：外部Agent分支与末尾均会引用 + + + + + + + + external_handled = False # 是否已由外部Agent(Dify/n8n/Coze)远程处理 + + + + + + + + + + + + + + + + # ===== 外部 Agent 远程调用分支 (Dify / n8n / Coze) ===== + + + + + + + + # 若本轮对话关联了带有效 agent_endpoint 的外部 Agent，则直接远程调用其平台， + + + + + + + + # 跳过本地 LLM「扮演」逻辑。effective_agent_ids 已在前面从请求/上下文继承。 + + + + + + + + external_agents = _resolve_market_agents(db, effective_agent_ids, user_id) + + + + + + + + external_agent = next((a for a in external_agents if a.get("agent_endpoint")), None) + + + + # ===== 3.3 多 Agent 编排(图引擎)快路 ===== + + _orchestrator = next((a for a in external_agents if a.get("orchestration")), None) + + if _orchestrator: + + try: + + from app.core.orchestrator import run_graph + + _answer, _trace = await run_graph( + + graph=_orchestrator["orchestration"], + + query=req.content + attachment_block, + + db=db, + + user_id=user_id, + + conv_id=conv_id, + + trace_id=get_trace_id(), + + ) + + ai_content = _answer or "编排未产生内容。" + + logger.info( + + f"[orchestrator] 图执行完成 trace_id={get_trace_id()} nodes={len(_trace.get('nodes', []))}" + + ) + + except Exception as e: + + logger.warning(f"[orchestrator] 图执行失败: {e}") + + ai_content = f"⚠️ 多 Agent 编排执行失败：{str(e)[:200]}" + + external_handled = True + + external_agent = None # 阻止下方外部 Agent 分支重复执行 + + + + + + + + if external_agent: + + + + + + + + agents = external_agents + + + + + + + + # 组装最近历史(供 n8n 等无平台会话态的服务维持多轮上下文) + + + + + + + + _history = _build_chat_history(db, conv_id, max_tokens=4000, exclude_latest_user=True) + + + + + + + + try: + + + + + + + + answer, new_ext_id = await _call_external_agent( + + + + + + + + db, external_agent, + + + + + + + + query=req.content + attachment_block, + + + + + + + + user_id=user_id, + + + + + + + + ext_conversation_id=conv.ext_agent_conversation_id, + + + + + + + + history=_history, + + + + + + + + ) + + + + + + + + ai_content = answer or "外部 Agent 未返回内容。" + + + + + + + + if new_ext_id: + + + + + + + + conv.ext_agent_conversation_id = new_ext_id + + + + + + + + except Exception as e: + + + + + + + + logger.warning( + + + + + + + + f"[external_agent] 远程调用失败 platform={external_agent.get('platform')} " + + + + + + + + f"endpoint={external_agent.get('agent_endpoint')}: {e}" + + + + + + + + ) + + + + + + + + ai_content = f"⚠️ 外部 Agent 调用失败：{str(e)[:200]}" + + + + + + + + external_handled = True + + + + + + + + + + + + + + + + # 解析模型：支持 Auto 模式（model_id='auto' 或未指定时，按能力自动匹配平台默认模型） + + + + + + + + if not external_handled: + + + + + + + + if model_id and model_id != "auto": + + + + + + + + ai_model = db.query(AIModel).filter(AIModel.id == model_id).first() + + + + + + + + if not ai_model: + + + + + + + + # Auto / 未指定 → 按用户输入意图匹配对应能力的默认模型，否则回退文本对话 + + + + + + + + intent = detect_media_intent(req.content) + + + + + + + + resolved = resolve_auto_model(intent, db) if intent else None + + + + + + + + ai_model = resolved or resolve_auto_model("text", db) + + + + + + + + + + + + + + + + + + + + # ===== Model Failover: Auto 模式下自动切换备选模型 ===== + _try_models = [ai_model] if (ai_model and ai_model.is_enabled) else [] + if not model_id or model_id == "auto": + # Auto 模式：附加其他已启用的文本模型作为备选 + _fallbacks = db.query(AIModel).filter( + AIModel.is_enabled == True, + AIModel.id != (ai_model.id if ai_model else "") + ).order_by(AIModel.is_default.desc()).limit(5).all() + _try_models.extend([m for m in _fallbacks if m not in _try_models]) + _retry_log = [] + _all_failed = False + + for _try_model in _try_models: + ai_model = _try_model + if ai_model: + + + + + + + + # ====== 图像/视频生成模型分支 ====== + + + + + + + + caps = [c.lower() for c in (ai_model.capabilities or [])] if ai_model else [] + + + + + + + + media_generating = False + + + + + + + + if ai_model and ("image" in caps or "video" in caps): + + + + + + + + # 异步生成：立即返回占位内容，后台任务轮询 AGNES 并回写， + + + + + + + + # 避免最长 5 分钟的视频轮询阻塞 HTTP 请求导致客户端超时断开。 + + + + + + + + media_generating = True + + + + + + + + if "video" in caps: + + + + + + + + ai_content = "⏳ 视频生成中…（通常 1~3 分钟，完成后自动刷新）" + + + + + + + + else: + + + + + + + + ai_content = "⏳ 图像生成中…（完成后自动刷新）" + + + + + + + + elif ai_model and ai_model.endpoint: + + + + + + + + provider = db.query(ModelProvider).filter(ModelProvider.id == ai_model.provider_id).first() + + + + + + + + api_key = ai_model.api_key or (provider.api_key if provider else None) + + + + + + + + api_url = ai_model.endpoint + + + + + + + + + + + + + + + + if not api_url and provider and provider.api_base_url: + + + + + + + + api_url = f"{provider.api_base_url.rstrip('/')}/v1/chat/completions" + + + + + + + + + + + + + + + + if api_key and api_url: + + + + + + + + try: + + + + + + + + # Stage A+B: 解析Skills和Agents + + + + + + + + skills = _resolve_market_skills(db, effective_skill_ids, user_id) + + + + + + + + agents = _resolve_market_agents(db, effective_agent_ids, user_id) + + + + + + + + mcps = _resolve_active_mcp_servers(db, effective_mcp_ids) + + + + + + + + # MCP 工具调用链路：拉取各 MCP 工具并转为 function definitions + + + + mcp_servers_conn = mcps + + + + mcp_function_definitions = [] + + + + mcp_tool_map = {} + + + + mcp_idx = 0 + + + + for _srv in mcp_servers_conn: + + + + 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: + + + + logger.warning(f"MCP 拉取工具失败 [{_srv.get('name')}]: {_e}") + + + + _tools = [] + + + + if _tools: + + + + try: + + + + _upd = db.query(MCPServer).filter(MCPServer.id == _srv["id"]).first() + + + + if _upd: + + + + _upd.tools_count = len(_tools) + + + + db.commit() + + + + except Exception: + + + + pass + + + + for _t in _tools: + + + + _fname = f"mcp_tool_{mcp_idx}" + + + + mcp_tool_map[_fname] = { + + + + "server_id": _srv["id"], + + + + "name": _t.get("name"), + + + + "server_name": _srv["name"], + + + + } + + + + _schema = _t.get("inputSchema") or {"type": "object", "properties": {}} + + + + if not isinstance(_schema, dict) or _schema.get("type") != "object": + + + + _schema = {"type": "object", "properties": {}} + + + + mcp_function_definitions.append({ + + + + "name": _fname, + + + + "description": f"[MCP·{_srv['name']}] {_t.get('description', '')}", + + + + "parameters": _schema, + + + + }) + + + + mcp_idx += 1 + + + + if mcp_function_definitions: + + + + effective_use_tools = True + + + + + + + + + + + + + + + + # Issue 8: 默认技能自动发现（仅当用户未显式选择技能时启用） + + + + + + + + # 每轮先寻找本账号已安装且平台已部署的技能，多个适用时由模型自行判断选 1 个最优。 + + + + + + + + used_auto_skills = False + + + + + + + + if not effective_skill_ids: + + + + + + + + try: + + + + + + + + discovered = _discover_installed_skills(db, user_id) + + + + + + + + except Exception as e: + + + + + + + + logger.warning(f"自动发现已安装技能失败: {e}") + + + + + + + + discovered = [] + + + + + + + + if discovered: + + + + + + + + skills = discovered + + + + + + + + used_auto_skills = True + + + + + + + + effective_use_tools = True + + + + + + + + + + + + + + + + # Stage B.1: Agent可能自带skill_ids和kb_ids + + + + + + + + agent_extra_skill_ids = [] + + + + + + + + agent_extra_kb_ids = [] + + + + + + + + agent_model_override = None + + + + + + + + for agent in agents: + + + + + + + + if agent.get("skill_ids"): + + + + + + + + agent_extra_skill_ids.extend(agent["skill_ids"]) + + + + + + + + if agent.get("knowledge_base_ids"): + + + + + + + + agent_extra_kb_ids.extend(agent["knowledge_base_ids"]) + + + + + + + + if agent.get("model_id"): + + + + + + + + agent_model_override = agent["model_id"] + + + + + + + + + + + + + + + + # 合并Agent自带的skills和KBs + + + + + + + + all_skill_ids = list(set(effective_skill_ids + agent_extra_skill_ids)) + + + + + + + + all_kb_ids = list(set(effective_kb_ids + agent_extra_kb_ids)) + + + + + + + + + + + + + + + + if agent_extra_skill_ids or agent_extra_kb_ids: + + + + + + + + # 重新解析合并后的skills + + + + + + + + skills = _resolve_market_skills(db, all_skill_ids, user_id) + + + + + + + + effective_use_tools = True + + + + + + + + + + + + + + + + # Stage C: 增强版KB检索 + + + + + + + + kb_context = await _retrieve_kb_context(db, all_kb_ids, req.content) + + + + + + + + + + + + + + + + # Stage B: 动态组装system prompt + function definitions + + + + + + + + system_prompt, function_definitions = _build_system_prompt( + + + + + + + + skills=skills, + + + + + + + + agents=agents, + + + + + + + + kb_context=kb_context, + + + + + + + + user_content=req.content, + + + + + + + + mcp=mcps, + + + + mcp_functions=mcp_function_definitions, + + + + auto_discovered=used_auto_skills, + + all_mcp=_list_enabled_mcp_servers(db), + + db=db, + + ) + + + + + + + + + + + + + + + + # 构建消息列表 + + + + + + + + # 实际调用追踪（Issue：反馈真实使用的技能/MCP，而非注入的全部） + + called_skill_names = [] + + called_mcp_names = [] + + + + messages = [{"role": "system", "content": system_prompt}] + + + + + + + + + + + + + + + + # 分层上下文引擎：预算感知历史 + 压缩摘要(若有) + + messages.extend(_build_chat_history(db, conv_id, max_tokens=6000, exclude_latest_user=True)) + + + + + + + + + + + + + + + + # Issue 1: 必须把附件内容(attachment_block)一并发给模型， + + + + + + + + # 否则用户在对话窗上传文件后模型“看不到”文件内容。 + + + + + + + + # Issue 2 修复：用户输入含 URL 且要求读取/加入/安装时，服务端抓取真实页面内容注入上下文 + try: + from app.api.url_reader import fetch_url_context + _url_ctx = await fetch_url_context(req.content) + except Exception as _e: + import logging as _logging + _logging.getLogger(__name__).warning("URL 上下文抓取失败(已跳过): %s", _e) + _url_ctx = None + if _url_ctx and _url_ctx.get("text"): + messages.append({ + "role": "system", + "content": ( + "【用户随消息提供的参考链接的真实内容】\n" + f"链接地址: {_url_ctx.get('url')}\n" + "下面是该链接页面的真实内容，请严格基于它来回答用户的需求" + "（例如按其『Quick Start / 加入步骤』逐步引导用户完成安装或加入），" + "不要凭空编造步骤：\n\n" + _url_ctx["text"] + ), + }) + process_steps.append({ + "type": "skill", + "name": "网页/仓库读取", + "result": f"已读取 {_url_ctx.get('url')}（{len(_url_ctx['text'])} 字符）", + }) + messages.append({"role": "user", "content": req.content + attachment_block}) + + + + + + + + + + + + + + + + # Stage B.3: Function Calling循环(最多5轮) + + + + + + + + max_tool_rounds = 5 + + + + + + + + # ===== 3.1 tool_policy 沙箱：聚合策略 + 初始化预算计数器 ===== + + _agent_tool_policy = _resolve_agent_tool_policy(agents) + + _tool_call_count = 0 + + for round_num in range(max_tool_rounds): + + _round_tool_count = 0 + + + + + + + + request_body = { + + + + + + + + "model": ai_model.name, + + + + + + + + "messages": messages, + + + + + + + + "max_tokens": 2048, + + + + + + + + } + + + + + + + + if function_definitions and effective_use_tools: + + + + + + + + request_body["tools"] = [ + + + + + + + + {"type": "function", "function": fd} + + + + + + + + for fd in function_definitions + + + + + + + + ] + + + + + + + + request_body["tool_choice"] = "auto" + + + + + + + + + + + + + + + + async with span("llm.completion", model=ai_model.id if ai_model else None): + try: + data = await achat_complete( + api_url, + api_key, + request_body, + proxy=_proxy_for_url(api_url), + timeout=ai_model.timeout or 120, + ) + except Exception as _exc: + if round_num == 0: + ai_content = f"AI模型调用失败 ({type(_exc).__name__}: {str(_exc)[:120]}), 请检查模型配置。" + logger.warning(f"[chat] 模型 {ai_model.name} 调用异常: {_exc}") + break + choice = data.get("choices", [{}])[0] + message = choice.get("message", {}) + + + + + + + + + + + + + + + + # 记录token使用量 + + + + + + + + usage = data.get("usage", {}) + + + + + + + + prompt_tokens += usage.get("prompt_tokens", 0) + + + + + + + + completion_tokens += usage.get("completion_tokens", 0) + + + + + + + + total_tokens += usage.get("total_tokens", 0) + + + + + + + + + + + + + + + + # 检查是否有tool_calls + + + + + + + + tool_calls = message.get("tool_calls", []) + # 过程步骤：捕获模型思考/推理(产生工具调用前的推理) + _think = (message.get("content") or "").strip() + if _think and tool_calls: + process_steps.append({ + "type": "thinking", + "text": _think, + }) + + + + + + + + + if tool_calls and effective_use_tools: + + + + + + + + # 将assistant消息(含tool_calls)加入上下文 + + + + + + + + messages.append({ + + + + + + + + "role": "assistant", + + + + + + + + "content": message.get("content") or "", + + + + + + + + "tool_calls": tool_calls, + + + + + + + + }) + + + + + + + + + + + + + + + + # Stage A.2: 真实Skill执行 + + + + + + + + for tc in tool_calls: + + + + + + + + func_name = tc.get("function", {}).get("name", "") + + # ===== 3.1 tool_policy 沙箱：逐工具校验（白名单 + 预算 + 每轮上限） ===== + + _mcp_info = mcp_tool_map.get(func_name) if func_name.startswith("mcp_tool_") else None + + _policy_check = _check_tool_policy(func_name, _mcp_info, skills, _agent_tool_policy, _tool_call_count, _round_tool_count) + + if not _policy_check["allowed"]: + + _reason = _policy_check["reason"] + + logger.warning( + + f"[tool_policy] blocked tool call: agent={_agent_tool_policy.get('agent_id') if _agent_tool_policy else None} " + + f"tool={func_name} reason={_reason} trace={get_trace_id()}" + + ) + + tool_result = f"[工具调用已被策略拦截] {_policy_check.get('message', _reason)}" + + messages.append({ + + "role": "tool", + + "tool_call_id": tc.get("id", f"call_{round_num}"), + + "content": tool_result, + + }) + + _tool_call_count += 1 + + _round_tool_count += 1 + + continue + + _tool_call_count += 1 + + _round_tool_count += 1 + + + + + + + + func_args_str = tc.get("function", {}).get("arguments", "{}") + + + + + + + + try: + + + + + + + + func_args = json.loads(func_args_str) + + + + + + + + except Exception: + + + + + + + + func_args = {"input": func_args_str} + + + + + + + + + + + + + + + + if func_name.startswith("mcp_tool_"): + + + + # MCP 工具调用：代理到对应 MCP 服务执行 tools/call + + + + _info = mcp_tool_map.get(func_name) + + + + tool_result = "[MCP 工具未找到或映射缺失]" + + + + if _info: + + + + _srv = next( + + + + (s for s in mcp_servers_conn if s["id"] == _info["server_id"]), + + + + None, + + + + ) + + + + if _srv: + + + + async with span(f"mcp.call:{_info.get('name') if isinstance(_info, dict) else None}"): + + tool_result = await call_mcp_tool( + + + + _srv, _info["name"], func_args + + + + ) + + + + called_mcp_names.append(_srv.get("name", "")) + process_steps.append({ + "type": "mcp", + "name": _info.get("name") if isinstance(_info, dict) else func_name, + "server": _srv.get("name", ""), + "args": func_args, + "result": (tool_result or "")[:800], + }) + + + + + else: + + + + async with span(f"skill.execute:{func_name}"): + + tool_result = _execute_skill_tool( + + + + db, func_name, func_args, skills, req.content + + + + ) + + + + _sk = next((s for s in skills if s.get("func_name") == func_name), None) + + + + if _sk: + + + + called_skill_names.append(_sk.get("name", "")) + process_steps.append({ + "type": "skill", + "name": (_sk.get("name") if _sk else func_name), + "args": func_args, + "result": (tool_result or "")[:800], + }) + + + + + + + + + messages.append({ + + + + + + + + "role": "tool", + + + + + + + + "tool_call_id": tc.get("id", f"call_{round_num}"), + + + + + + + + "content": tool_result, + + + + + + + + }) + + + + + + + + + + + + + + + + # 继续下一轮让模型处理tool结果 + + + + + + + + continue + + + + + + + + else: + + + + + + + + # 普通回复 + + + + + + + + ai_content = message.get("content", "") + + + + + + + + if not ai_content and round_num > 0: + + + + + + + + ai_content = "AI模型已处理工具调用结果, 但返回了空响应。" + + + + + + + + break + + + + + + + + + + + + + + + + if not ai_content: + + + + + + + + ai_content = "AI模型返回了空响应, 请稍后重试。" + + + + + + + + + + + + + + + + except Exception as e: + + + + + + + + logger.warning(f"AI API调用异常: {e}") + + + + + + + + ai_content = "AI服务暂时不可用, 请稍后重试或检查模型API配置。" + + + + + + + + else: + + + + + + + + ai_content = "该模型缺少API密钥或端点配置, 请在后台管理-大模型库中配置API Key和Endpoint。" + + + + + + + + else: + + + + + + + + ai_content = "请选择一个已配置的AI模型进行对话。" + + + + + + + + elif not external_handled: + + + + + + + + ai_content = "请先在后台管理-大模型库配置默认对话模型（或聊天中选择模型），Auto 模式下将自动使用平台默认模型。" + + + + + # ===== Post-check: 失败则切换备选，成功则跳出 ===== + _is_error = ( + ai_content and ( + "AI模型调用失败" in str(ai_content) + or "AI服务暂时不可用" in str(ai_content) + or "AI模型返回了空响应" in str(ai_content) + or "该模型缺少API密钥" in str(ai_content) + ) + ) + if _is_error: + _retry_log.append(f"{ai_model.name}: {str(ai_content)[:60]}") + logger.warning( + f"[Failover] 模型 {ai_model.name} 失败，切换至备选..." + ) + continue # try next model + + # Success — exit failover loop + break + else: + # Loop completed without break = all models failed + _all_failed = True + + # ===== All models failed? ===== + if _all_failed: + _names = ", ".join([m.name for m in _try_models]) + _details = "; ".join(_retry_log[-3:]) if _retry_log else "未知错误" + ai_content = ( + f"已尝试 {len(_try_models)} 个模型（{_names}）均调用失败。\n" + f"最后错误: {_details}\n" + f"请稍后重试或检查模型配置。" + ) + logger.error(f"[Failover] 全部 {len(_try_models)} 个模型均失败: {_details}") + + + + + + + + + + + + + + + + + + + + + ai_token_count = len(ai_content) // 2 + + + + + + + + ai_msg = Message( + + + + + + + + id=gen_uuid(), + + + + + + + + conversation_id=conv_id, + + + + + + + + role="assistant", + + + + + + + + content=ai_content, + + + + + + + + token_used=ai_token_count, + + + + + + + + model_id=ai_model.id if ai_model else (req.model_id or conv.model_id), + + + + + + + + skill_ids=req.skill_ids if req.skill_ids else None, + + + + + + + + agent_ids=req.agent_ids if req.agent_ids else None, + + + + + + + + knowledge_ids=req.knowledge_base_ids if req.knowledge_base_ids else None, + + + + + + + + created_at=cst_now(), + + + + + + + + ) + + + + + + + + db.add(ai_msg) + + + + + + + + + + + + + + + + # 后台异步：将实质性文本回答入库到 Dify 知识库（量化后可供 RAG 复用） + + + + + + + + _ingest_text = (ai_content or "").strip() + + + + + + + + if len(_ingest_text) > 120 and not ( + + + + + + + + _ingest_text.startswith("[MEDIA") + + + + + + + + or _ingest_text.startswith("data:") + + + + + + + + or _ingest_text.startswith("http") + + + + + + + + or " 反馈配对的技能名 + + + + + + + + try: + + + + + + + + _sel = _resolve_market_skills(db, effective_skill_ids, user_id) + + + + + + + + used_skills = [s["name"] for s in _sel] + + + + + + + + except Exception: + + + + + + + + used_skills = [] + + + + + + + + elif called_skill_names: + + + + + + + + # 未显式选技能 -> 反馈 AI 本轮实际调用的技能（可能为空） + + + + + + + + used_skills = list(dict.fromkeys(called_skill_names)) + + + + + + + + if effective_mcp_ids: + + + + + + + + # 用户显式选择了 MCP -> 反馈配对的 MCP 名 + + + + + + + + try: + + + + + + + + _sel_mcp = _resolve_active_mcp_servers(db, effective_mcp_ids) + + + + + + + + used_mcp = [m["name"] for m in _sel_mcp] + + + + + + + + except Exception: + + + + + + + + used_mcp = [] + + + + + + + + elif called_mcp_names: + + + + + + + + used_mcp = list(dict.fromkeys(called_mcp_names)) + + + + + + + + ai_msg.used_skills = used_skills + + + + + + + + ai_msg.used_mcp = used_mcp + + + + + + + + + + + + + + + + # 更新对话统计 + + + + + + + + # 注意：knowledge_graph_count 仅应在用户实际点击「转知识图谱」时 +1， + # 不可在每次 AI 回复时无条件自增，否则未操作也会累积计数（见 #715）。 + + + + + + + + conv.updated_at = cst_now() + + + + + + + + + + + + + + + + # 记录Token使用（媒体生成走 _generate_media_task 异步补写真实消耗，此处跳过避免占位文本重复记账） + + + + + + + + if not media_generating: + + + + + + + + token_usage = TokenUsageLog( + + + + + + + + id=gen_uuid(), + + + + + + + + user_id=current_user.id, + + + + + + + + model_id=ai_model.id if ai_model else (req.model_id or conv.model_id), + + + + + + + + model_name=ai_model.name if ai_model else None, + + + + + + + + conversation_id=conv_id, + + trace_id=get_trace_id(), + + + + + + + + + + prompt_tokens=prompt_tokens or len(req.content) // 2, + + + + + + + + completion_tokens=completion_tokens or ai_token_count, + + + + + + + + total_tokens=total_tokens or (len(req.content) // 2 + ai_token_count), + + + + + + + + created_at=cst_now(), + + + + + + + + ) + + + + + + + + db.add(token_usage) + + + + + + + + + + + + + + + + # 进化轨迹记录 + + + + + + + + try: + + + + + + + + existing = db.query(EvolutionTrajectory).filter( + + + + + + + + EvolutionTrajectory.conversation_id == conv_id + + + + + + + + ).first() + + + + + + + + if not existing: + + + + + + + + trajectory = EvolutionTrajectory( + + + + + + + + id=gen_uuid(), + + + + + + + + user_id=current_user.id, + + + + + + + + conversation_id=conv_id, + + + + + + + + status="discovered", + + + + + + + + token_used=ai_token_count, + + + + + + + + created_at=cst_now(), + + + + + + + + ) + + + + + + + + db.add(trajectory) + + + + + + + + db.flush() + + + + + + + + + + + + + + + + try: + + + + + + + + from app.services.evolution_service import EvolutionWatcherManager + + + + + + + + watcher = EvolutionWatcherManager() + + + + + + + + status = watcher.get_user_status(current_user.id, lambda: SessionLocal()) + + + + + + + + if not status.get("scheduler_running"): + + + + + + + + watcher.start_user_evolution(current_user.id, lambda: SessionLocal()) + + + + + + + + logger.info(f"[Evolution] 自动启动用户 {current_user.id} 的进化Pipeline") + + + + + + + + except Exception as evolution_err: + + + + + + + + logger.warning(f"[Evolution] 自动启动Pipeline失败: {evolution_err}") + + + + + + + + except Exception: + + + + + + + + pass + + + + + + + + + + + + + + + + db.commit() + + + + + + + + # 后台异步压缩历史(分层记忆)，不阻塞本次响应 + + asyncio.create_task(_maybe_compress_history(conv_id)) + + + + db.refresh(user_msg) + + + + + + + + db.refresh(ai_msg) + + + + + + + + + + + + + + + + # 媒体生成改为后台异步：立即返回占位，后台任务完成后回写消息内容 + + + + + + + + if media_generating: + + + + + + + + try: + + + + + + + + asyncio.create_task( + + + + + + + + _generate_media_task(ai_msg.id, ai_model.id, req.content, current_user.id) + + + + + + + + ) + + + + + + + + except Exception as e: + + + + + + + + logger.warning(f"[_generate_media_task] 启动后台任务失败: {e}", exc_info=True) + # 启动失败兜底：不让占位消息永久停在「⏳ 生成中…」 + try: + ai_msg.content = "⚠️ 媒体生成启动失败，请稍后重试" + db.commit() + except Exception: + pass + + + + + + + + + + + + + + + + return { + + + + + + + + "code": 0, + + + + + + + + "data": { + + + + + + + + "user_message": { + + + + + + + + "id": user_msg.id, "role": user_msg.role, "content": user_msg.content, + + + + + + + + "token_used": user_msg.token_used, "created_at": str(user_msg.created_at), + + + + + + + + }, + + + + + + + + "assistant_message": { + + + + + + + + "id": ai_msg.id, "role": ai_msg.role, "content": ai_msg.content, + + + + + + + + "token_used": ai_msg.token_used, "created_at": str(ai_msg.created_at), + + + + + + + + "used_skills": used_skills, + + "used_mcp": used_mcp, + + + + + + + + "generating": media_generating, + "process_steps": process_steps, + + + + + + + + + }, + + + + + + + + }, + + + + + + + + "message": "发送成功", + + + + + + + + } + + + + + +# ====================================================================== + +# 3.1 tool_policy 沙箱辅助函数 + +# ====================================================================== + +def _resolve_agent_tool_policy(agents): + + """从已解析的 agents 聚合工具策略。多 agent 时取首个带有非空策略者。""" + + if not agents: + + return None + + for ag in agents: + + tp = ag.get("tool_policy") + + if tp and isinstance(tp, dict): + + tp = dict(tp) + + tp["agent_id"] = ag.get("id") + + return tp + + return None + + + + + +def _check_tool_policy(func_name, mcp_info, skills, policy, call_count, round_count): + + """校验单个工具调用是否被策略允许。 + + + + 返回 {"allowed": bool, "reason": str, "tool_id": str|None, "message": str} + + policy 为 None/空 -> 允许（向后兼容）。 + + """ + + if not policy: + + return {"allowed": True, "tool_id": None} + + allowed = policy.get("allowed_tools") or {} + + skill_ids = allowed.get("skill_ids") or [] + + mcp_ids = allowed.get("mcp_ids") or [] + + is_mcp = func_name.startswith("mcp_tool_") + + tool_id = None + + if is_mcp: + + if mcp_info and isinstance(mcp_info, dict): + + tool_id = mcp_info.get("server_id") + + if mcp_ids and tool_id not in mcp_ids: + + return {"allowed": False, "reason": "mcp_not_in_allowlist", "tool_id": tool_id, + + "message": "MCP 服务不在该智能体的允许列表中"} + + else: + + if skills: + + _s = next((s for s in skills if s.get("func_name") == func_name), None) + + if _s: + + tool_id = _s.get("id") + + if skill_ids and tool_id not in skill_ids: + + return {"allowed": False, "reason": "skill_not_in_allowlist", "tool_id": tool_id, + + "message": "技能不在该智能体的允许列表中"} + + budget = policy.get("max_calls_per_conversation") + + if budget is not None and call_count >= int(budget): + + return {"allowed": False, "reason": "budget_exceeded", "tool_id": tool_id, + + "message": policy.get("fallback_message") + + or "该智能体已达到工具调用上限，请开启新对话或调整智能体设置。"} + + per_round = policy.get("per_round_cap") + + if per_round is not None and round_count >= int(per_round): + + return {"allowed": False, "reason": "per_round_cap", "tool_id": tool_id, + + "message": "本轮工具调用已达上限"} + + return {"allowed": True, "tool_id": tool_id} + diff --git a/backend/python/app/api/im_llm.py b/backend/python/app/api/im_llm.py index f528dcc..bbbc4b4 100644 --- a/backend/python/app/api/im_llm.py +++ b/backend/python/app/api/im_llm.py @@ -1,283 +1,282 @@ -""" -IM Bot 真实 LLM 调用。 - -原实现位于 channels.py（通用频道管理模块），该模块已移除。 -为保持 IM Bot（微信机器人等）正常生成回复，将 _call_real_llm 迁移至此独立模块。 - -行为（对齐 AI 聊天 — 2026-07-23 同步至最新）： -- Issue 7：图像/视频/语音意图优先走平台媒体模型（与 AI 聊天同一套 Auto 路由）， - 不再对"制作图片"这类请求只回文本。 -- Issue 8：默认 Auto 模型；每轮先自动发现本账号已安装且平台已部署的技能， - 多个适用时由模型自行判断选 1 个最优，通过 function calling 执行。 -- 2026-07-15：修复 trust_env=False + proxy 处理，避免容器坏代理导致调用失败； - 媒体生成前先经 _optimize_media_prompt 精炼提示词，提升 AGNES 生成质量。 -- 2026-07-23：对齐 Web Chat 最新能力： - * 模型 failover（Auto 模式自动切换备选模型） - * 平台知识库注入（get_platform_knowledge，技能清单实时同步） -""" -import asyncio -import json -import logging -import re as _re -from datetime import datetime, timezone, timedelta -from typing import Optional -from urllib.parse import urlparse as _urlparse - -import httpx -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__) - - -# 香港出网代理(实测对 AGNES 延迟更低，2026-07-15 用户实测)，与 chat.py 同值。 -_HK_PROXY = "http://qyyfw:XHR521yhf@103.242.14.208:11495" - - -def _proxy_for_url(url: str) -> str | None: - """按目标域名选择是否走代理（与 chat.py 同逻辑）。 - - 2026-07-15 调整：AGNES 改走香港代理(实测延迟更低)，不再直连。 - 国内域名(尊重 NO_PROXY)直连，更稳定。 - """ - import os - host = (_urlparse(url).hostname or "").lower() - if "agnes-ai.com" in host or "agnes-ai.space" in host: - return _HK_PROXY - if "qihuyun.com" in host: - return None - _no_proxy = [h.strip().lower() for h in os.environ.get("NO_PROXY", "").split(",") if h.strip()] - for _np in _no_proxy: - if host == _np or host.endswith("." + _np) or _np in host: - return None - return os.environ.get("HTTPS_PROXY") or os.environ.get("HTTP_PROXY") or "http://172.18.0.1:7890" - - -def _run_async(coro, timeout=None): - """在同步上下文中运行协程，兼容「无事件循环」与「已有事件循环」两种场景。 - - IM 个人微信在独立线程中调用（无事件循环）→ asyncio.run； - IM 多平台 Webhook 可能在 FastAPI 协程内调用（已有事件循环）→ 丢到线程池执行。 - - timeout: 可选，秒数；超时抛出 asyncio.TimeoutError（由调用方捕获）。 - """ - try: - loop = asyncio.get_running_loop() - except RuntimeError: - loop = None - - if timeout is not None: - async def _timed(): - return await asyncio.wait_for(coro, timeout=timeout) - target = _timed() - else: - target = coro - - if loop is not None and loop.is_running(): - import concurrent.futures - with concurrent.futures.ThreadPoolExecutor(1) as ex: - return ex.submit(lambda: asyncio.run(target)).result(timeout=(timeout + 10) if timeout else None) - return asyncio.run(target) - - -def _log_media_usage(db: Session, user_id: str, ai_model, media_type: str, prompt: str): - """IM Bot 媒体生成记账：把图片/视频消耗写入 TokenUsageLog，供驾驶舱统计。 - - 此前 IM Bot 路径完全不写 TokenUsageLog，导致微信生成的图片/视频在驾驶舱里"查无此项"。 - 媒体 API 不返回 token 用量，这里用与 chat.py _generate_media_task 一致的估算。 - """ - if not user_id or not ai_model: - return - try: - base = 1200 if media_type == "video" else 400 - est = base + (len(prompt or "") // 4) - db.add(TokenUsageLog( - id=gen_uuid(), - 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, - created_at=datetime.now(timezone(timedelta(hours=8))).replace(tzinfo=None), - )) - db.commit() - except Exception as e: - try: - db.rollback() - except Exception: - pass - logger.warning(f"[im_llm] 媒体 Token 记账失败(忽略): {e}") - - -def _call_real_llm(db: Session, user=None, prompt: str = "", platform: str = "im", user_id: str = None) -> "str | dict": - """调用七虎AI 生成回复，供 IM Bot 复用。支持媒体意图路由、已安装技能自动发现、模型 failover。""" - try: - from app.api.chat import ( - _generate_media, - detect_media_intent, - _discover_installed_skills, - _build_system_prompt, - _execute_skill_tool, - ) - - # ===== 1) 媒体意图路由（Issue 7）===== - intent = detect_media_intent(prompt) - if intent: - ai_model = resolve_auto_model(intent, db) - if ai_model: - try: - result = _run_async(_generate_media(db, ai_model, prompt, user_id), timeout=300) - except Exception: - # 任何未预期异常(如 AGNES 连接重置)统一兜底为友好提示， - # 绝不直接把原始异常(如 "Server disconnected")甩给用户 - result = "⚠️ 媒体生成暂时失败（连接或服务波动），请稍后重试。" - if result: - # 媒体生成成功：返回结构化对象，交由 im_bot_user 在微信内联发送图片/视频 - s = result.strip() - if s.startswith(" str | None: + """按目标域名选择是否走代理（与 chat.py 同逻辑）。 + + 2026-07-15 调整：AGNES 改走香港代理(实测延迟更低)，不再直连。 + 国内域名(尊重 NO_PROXY)直连，更稳定。 + """ + import os + host = (_urlparse(url).hostname or "").lower() + if "agnes-ai.com" in host or "agnes-ai.space" in host: + return _HK_PROXY + if "qihuyun.com" in host: + return None + _no_proxy = [h.strip().lower() for h in os.environ.get("NO_PROXY", "").split(",") if h.strip()] + for _np in _no_proxy: + if host == _np or host.endswith("." + _np) or _np in host: + return None + return os.environ.get("HTTPS_PROXY") or os.environ.get("HTTP_PROXY") or "http://172.18.0.1:7890" + + +def _run_async(coro, timeout=None): + """在同步上下文中运行协程，兼容「无事件循环」与「已有事件循环」两种场景。 + + IM 个人微信在独立线程中调用（无事件循环）→ asyncio.run； + IM 多平台 Webhook 可能在 FastAPI 协程内调用（已有事件循环）→ 丢到线程池执行。 + + timeout: 可选，秒数；超时抛出 asyncio.TimeoutError（由调用方捕获）。 + """ + try: + loop = asyncio.get_running_loop() + except RuntimeError: + loop = None + + if timeout is not None: + async def _timed(): + return await asyncio.wait_for(coro, timeout=timeout) + target = _timed() + else: + target = coro + + if loop is not None and loop.is_running(): + import concurrent.futures + with concurrent.futures.ThreadPoolExecutor(1) as ex: + return ex.submit(lambda: asyncio.run(target)).result(timeout=(timeout + 10) if timeout else None) + return asyncio.run(target) + + +def _log_media_usage(db: Session, user_id: str, ai_model, media_type: str, prompt: str): + """IM Bot 媒体生成记账：把图片/视频消耗写入 TokenUsageLog，供驾驶舱统计。 + + 此前 IM Bot 路径完全不写 TokenUsageLog，导致微信生成的图片/视频在驾驶舱里"查无此项"。 + 媒体 API 不返回 token 用量，这里用与 chat.py _generate_media_task 一致的估算。 + """ + if not user_id or not ai_model: + return + try: + base = 1200 if media_type == "video" else 400 + est = base + (len(prompt or "") // 4) + db.add(TokenUsageLog( + id=gen_uuid(), + 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, + created_at=datetime.now(timezone(timedelta(hours=8))).replace(tzinfo=None), + )) + db.commit() + except Exception as e: + try: + db.rollback() + except Exception: + pass + logger.warning(f"[im_llm] 媒体 Token 记账失败(忽略): {e}") + + +def _call_real_llm(db: Session, user=None, prompt: str = "", platform: str = "im", user_id: str = None) -> "str | dict": + """调用七虎AI 生成回复，供 IM Bot 复用。支持媒体意图路由、已安装技能自动发现、模型 failover。""" + try: + from app.api.chat import ( + _generate_media, + detect_media_intent, + _discover_installed_skills, + _build_system_prompt, + _execute_skill_tool, + ) + + # ===== 1) 媒体意图路由（Issue 7）===== + intent = detect_media_intent(prompt) + if intent: + ai_model = resolve_auto_model(intent, db) + if ai_model: + try: + result = _run_async(_generate_media(db, ai_model, prompt, user_id), timeout=300) + except Exception: + # 任何未预期异常(如 AGNES 连接重置)统一兜底为友好提示， + # 绝不直接把原始异常(如 "Server disconnected")甩给用户 + result = "⚠️ 媒体生成暂时失败（连接或服务波动），请稍后重试。" + if result: + # 媒体生成成功：返回结构化对象，交由 im_bot_user 在微信内联发送图片/视频 + s = result.strip() + if s.startswith(" dict，形状对齐上游 `data` 响应： + {"choices":[{"message":{"content","tool_calls","reasoning_content"}}], "usage":{...}} + - stream=True -> async generator，逐块产出 ("content"|"reasoning", text) 增量 +""" +from __future__ import annotations + +import httpx +from openai import AsyncOpenAI, OpenAI + + +def _base_url(api_url: str) -> str: + """从完整 /v1/chat/completions 或 /v1/embeddings 地址推导 OpenAI SDK 所需的 base_url。""" + u = (api_url or "").rstrip() + for suf in ("/v1/chat/completions", "/chat/completions", "/v1/embeddings", "/embeddings"): + if u.endswith(suf): + return u[: -len(suf)] + return u + + +def build_openai_client(api_url, api_key, *, proxy=None, timeout=120) -> AsyncOpenAI: + base = _base_url(api_url) + # 始终 trust_env=False：避免容器 HTTP_PROXY 对国内端点(siliconflow 等)造成 502； + # 代理经 proxy 显式传入（AGNES 等），未传则直连。 + http_client = httpx.AsyncClient(proxy=proxy, trust_env=False) + return AsyncOpenAI( + base_url=base, + api_key=api_key or "EMPTY", + timeout=timeout, + http_client=http_client, + ) + + +def _create_kwargs(request_body: dict) -> dict: + kw = { + "model": request_body.get("model"), + "messages": request_body.get("messages", []), + "stream": False, + } + if request_body.get("temperature") is not None: + kw["temperature"] = request_body["temperature"] + if request_body.get("top_p") is not None: + kw["top_p"] = request_body["top_p"] + if request_body.get("max_tokens") is not None: + kw["max_tokens"] = request_body["max_tokens"] + if request_body.get("stop") is not None: + kw["stop"] = request_body["stop"] + tools = request_body.get("tools") + if tools: + kw["tools"] = tools + kw["tool_choice"] = request_body.get("tool_choice", "auto") + return kw + + +def _normalize_tool_calls(tcs): + if not tcs: + return None + out = [] + for tc in tcs: + out.append( + { + "id": getattr(tc, "id", None), + "type": "function", + "function": { + "name": getattr(tc.function, "name", None), + "arguments": getattr(tc.function, "arguments", "") or "", + }, + } + ) + return out + + +async def achat_complete(api_url, api_key, request_body, *, proxy=None, timeout=120, stream=False): + """统一的 OpenAI 兼容聊天补全。详见模块 docstring。""" + if stream: + + async def _gen(): + client = build_openai_client(api_url, api_key, proxy=proxy, timeout=timeout) + try: + gen = await client.chat.completions.create( + **_create_kwargs(request_body), stream=True + ) + async for chunk in gen: + if not chunk.choices: + continue + delta = chunk.choices[0].delta + rc = getattr(delta, "reasoning_content", None) + if rc: + yield ("reasoning", rc) + ct = getattr(delta, "content", None) + if ct: + yield ("content", ct) + finally: + await client.close() + + return _gen() + + client = build_openai_client(api_url, api_key, proxy=proxy, timeout=timeout) + async with client: + resp = await client.chat.completions.create(**_create_kwargs(request_body)) + msg = resp.choices[0].message + return { + "choices": [ + { + "message": { + "content": msg.content or "", + "tool_calls": _normalize_tool_calls(getattr(msg, "tool_calls", None)), + "reasoning_content": getattr(msg, "reasoning_content", None), + } + } + ], + "usage": (resp.usage.model_dump() if resp.usage else {}), + } + + +def build_openai_client_sync(api_url, api_key, *, proxy=None, timeout=120) -> OpenAI: + """同步客户端工厂（供 IM bot 等同步上下文复用同一标准模型层）。""" + base = _base_url(api_url) + http_client = httpx.Client(proxy=proxy, trust_env=False) + return OpenAI(base_url=base, api_key=api_key or "EMPTY", timeout=timeout, http_client=http_client) + + +def chat_complete(api_url, api_key, request_body, *, proxy=None, timeout=120): + """同步版（与 achat_complete 行为/返回形状完全一致），供非 async 上下文使用。""" + client = build_openai_client_sync(api_url, api_key, proxy=proxy, timeout=timeout) + with client: + resp = client.chat.completions.create(**_create_kwargs(request_body)) + msg = resp.choices[0].message + return { + "choices": [ + { + "message": { + "content": msg.content or "", + "tool_calls": _normalize_tool_calls(getattr(msg, "tool_calls", None)), + "reasoning_content": getattr(msg, "reasoning_content", None), + } + } + ], + "usage": (resp.usage.model_dump() if resp.usage else {}), + } + + +# --------------------------------------------------------------------------- +# 嵌入（Embeddings）：OpenAI 兼容 /v1/embeddings，收敛散落的裸 httpx 嵌入调用 +# --------------------------------------------------------------------------- +def _embed_kwargs(input, model): + return {"model": model, "input": input if isinstance(input, list) else [input]} + + +async def aembed(api_url, api_key, input, model, *, proxy=None, timeout=60) -> list: + """统一的 OpenAI 兼容文本嵌入（异步）。返回 list[list[float]]。 + + 供 pgvector_search.get_embedding 等 embeddings 路径复用，把散落的裸 httpx + 嵌入调用收敛到同一标准模型层（与聊天共用代理/超时/failover 语义）。 + """ + client = build_openai_client(api_url, api_key, proxy=proxy, timeout=timeout) + async with client: + resp = await client.embeddings.create(**_embed_kwargs(input, model)) + return [d.embedding for d in resp.data] + + +def embed(api_url, api_key, input, model, *, proxy=None, timeout=60) -> list: + """同步版嵌入（与非 async 上下文复用同一标准模型层）。""" + client = build_openai_client_sync(api_url, api_key, proxy=proxy, timeout=timeout) + with client: + resp = client.embeddings.create(**_embed_kwargs(input, model)) + return [d.embedding for d in resp.data] diff --git a/backend/python/app/core/observability.py b/backend/python/app/core/observability.py index 1bce417..a1521d4 100644 --- a/backend/python/app/core/observability.py +++ b/backend/python/app/core/observability.py @@ -1,6 +1,6 @@ """ -Qihu-AIFlow 统一可观测基座 (Phase 2.1) -===================================== +Qihu-AIFlow 统一可观测基座 (Phase 2.1 + Phase 1.I OTel 增强) +============================================================= 提供贯穿一次请求的全局链路 ID（trace_id）与结构化日志能力， 为后续重构（上下文引擎、媒体/LLM 成本归因）建立回归基线。 @@ -10,13 +10,17 @@ Qihu-AIFlow 统一可观测基座 (Phase 2.1) TraceIdFilter 自动注入，调用方无需改动。 - 不引入第三方依赖（structlog 未安装），自带可选的 JSON formatter。 +Phase 1.I 增强：span() 在保持原有日志计时的同时，额外嵌套一个标准 OpenTelemetry +span（经 app.core.otel 桥接，OTLP 导出到 Grafana）。OTel 失败时自动降级为 no-op， +绝不改变既有日志行为与请求链路。 + 使用方式： from app.core.observability import set_trace_id, get_trace_id, span # 在中间件里为每个请求注入： set_trace_id(request_id) - # 在关键调用处打计时 span（自动带 trace_id）： + # 在关键调用处打计时 span（自动带 trace_id + OTel span）： async with span("chat.send_message", conv_id=conv_id): return await _send_message_core(...) """ @@ -26,6 +30,8 @@ import uuid import json from contextvars import ContextVar +from app.core.otel import current_span_cm, record_span_metrics + # --------------------------------------------------------------------------- # trace_id：贯穿一次请求（含其内所有子调用/协程）的全局链路 ID # --------------------------------------------------------------------------- @@ -103,10 +109,10 @@ def get_logger(name: str | None = None) -> logging.Logger: # --------------------------------------------------------------------------- -# span：结构化计时上下文管理器（耗时 + trace_id + 自定义字段） +# span：结构化计时上下文管理器（耗时 + trace_id + 自定义字段 + OTel 嵌套） # --------------------------------------------------------------------------- class _Span: - """同步/异步通用的计时 span。退出时记录 elapsed_ms 与状态。""" + """同步/异步通用的计时 span。退出时记录 elapsed_ms 与状态，并结束嵌套的 OTel span。""" def __init__(self, name: str, **fields): self.name = name @@ -114,6 +120,7 @@ class _Span: self._logger = logging.getLogger("aiflow.span") self._start = None self.elapsed_ms = 0.0 + self._otel_cm = None # 嵌套的 OTel span 上下文管理器 def _enter(self): self._start = time.perf_counter() @@ -121,6 +128,10 @@ class _Span: f"SPAN_START {self.name}", extra={"span": self.name, "event": "start", "trace_id": get_trace_id(), **self.fields}, ) + # OTel 嵌套 span（失败安全：current_span_cm 返回 None 时 no-op） + self._otel_cm = current_span_cm(self.name, self.fields) + if self._otel_cm is not None: + self._otel_cm.__enter__() return self def _exit(self, exc_type, exc, tb): @@ -137,6 +148,24 @@ class _Span: **self.fields, }, ) + # 结束嵌套的 OTel span（失败安全） + if self._otel_cm is not None: + try: + from opentelemetry import trace as _ot + if exc_type is not None: + self._otel_cm.set_status( + _ot.trace.Status(_ot.trace.StatusCode.ERROR, str(exc_type.__name__)) + ) + else: + self._otel_cm.set_status(_ot.trace.Status(_ot.trace.StatusCode.OK)) + self._otel_cm.__exit__(exc_type, exc, tb) + except Exception: + pass + # 记录 span 指标（耗时直方图 + 计数），供 Grafana 出图（失败安全） + try: + record_span_metrics(self.name, self.elapsed_ms, status) + except Exception: + pass return False # 不吞掉异常 # 同步协议 diff --git a/backend/python/app/core/otel.py b/backend/python/app/core/otel.py new file mode 100644 index 0000000..cc723bf --- /dev/null +++ b/backend/python/app/core/otel.py @@ -0,0 +1,131 @@ +""" +OpenTelemetry 桥接 (Phase 1.I 标准化) +===================================== +设计基准（标准对照表见 STANDARDS_AUDIT_2026-07-24.md §I）： +- 标准：OpenTelemetry SDK + OTLP gRPC exporter —— 云原生可观测事实标准。 +- traces：span 经 observability.span() 嵌套导出（见 observability.py）。 +- metrics：MeterProvider 经 OTLP 导出 span 耗时直方图 + 调用计数，供 Grafana 出图。 +- 失败安全（生产红线）：若 OTLP 端点不可达 / 初始化异常，自动降级为 no-op， + 绝不阻断业务请求、不产生未捕获异常。 +- 门控：OTEL_ENABLED=1 默认开启；置 0 则零开销、不走 OTel。 +- 发射目标：OTEL_EXPORTER_OTLP_ENDPOINT，默认 http://qihu-aiflow-alloy:4317 + （经 Grafana Alloy 桥接 OTLP→Prometheus，再由 Prometheus 喂给 Grafana 出图）。 + +对外 API： + init_otel() 首次调用时构建 TracerProvider + MeterProvider（幂等） + get_tracer() 返回 tracer（未启用/失败则 None） + get_meter() 返回 meter（未启用/失败则 None） + current_span_cm(name) 返回一个 OTel span 的上下文管理器（供 observability 嵌套） + record_span_metrics(name, elapsed_ms, status) 记录 span 指标（耗时直方图 + 计数） +""" + +import os +import logging + +logger = logging.getLogger(__name__) + +_ENABLED = os.getenv("OTEL_ENABLED", "1") not in ("0", "false", "False", "") +_ENDPOINT = os.getenv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://qihu-aiflow-alloy:4317").rstrip("/") +if not _ENDPOINT.startswith("http://") and not _ENDPOINT.startswith("https://"): + _ENDPOINT = "http://" + _ENDPOINT + +_tracer = None +_meter = None +_initialized = False +_duration_hist = None +_span_counter = None + + +def init_otel() -> None: + """构建 TracerProvider + MeterProvider（幂等）。失败则降级为 no-op。""" + global _tracer, _meter, _initialized + if _initialized: + return + _initialized = True + if not _ENABLED: + logger.info("[otel] disabled (OTEL_ENABLED=0)") + return + try: + from opentelemetry import trace, metrics + from opentelemetry.sdk.resources import Resource + from opentelemetry.semconv.resource import ResourceAttributes + from opentelemetry.sdk.trace import TracerProvider + from opentelemetry.sdk.trace.export import BatchSpanProcessor + from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import ( + OTLPSpanExporter, + ) + from opentelemetry.sdk.metrics import MeterProvider + from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader + from opentelemetry.exporter.otlp.proto.grpc.metric_exporter import ( + OTLPMetricExporter, + ) + + resource = Resource.create({ResourceAttributes.SERVICE_NAME: "qihu-aiflow"}) + + # --- traces --- + trace_provider = TracerProvider(resource=resource) + trace_provider.add_span_processor( + BatchSpanProcessor(OTLPSpanExporter(endpoint=_ENDPOINT, insecure=True)) + ) + trace.set_tracer_provider(trace_provider) + _tracer = trace.get_tracer("qihu-aiflow") + + # --- metrics --- + metric_reader = PeriodicExportingMetricReader( + OTLPMetricExporter(endpoint=_ENDPOINT, insecure=True) + ) + meter_provider = MeterProvider(resource=resource, metric_readers=[metric_reader]) + metrics.set_meter_provider(meter_provider) + _meter = metrics.get_meter("qihu-aiflow") + + logger.info(f"[otel] tracer+meter ready -> {_ENDPOINT}") + except Exception as e: # pragma: no cover - 降级保护 + logger.warning(f"[otel] init failed, degraded to no-op: {e}") + _tracer = None + _meter = None + + +def get_tracer(): + """返回 tracer；首次调用触发 init。未启用/失败返回 None。""" + if _tracer is None and _ENABLED: + init_otel() + return _tracer + + +def get_meter(): + """返回 meter；首次调用触发 init。未启用/失败返回 None。""" + if _meter is None and _ENABLED: + init_otel() + return _meter + + +def current_span_cm(name: str, attributes: dict | None = None): + """返回一个 OTel span 的上下文管理器（同步/异步通用）。失败时返回 None。""" + tr = get_tracer() + if tr is None: + return None + try: + return tr.start_as_current_span(name, attributes=attributes or {}) + except Exception as e: # pragma: no cover + logger.warning(f"[otel] start_span failed: {e}") + return None + + +def record_span_metrics(name: str, elapsed_ms: float, status: str) -> None: + """记录 span 指标：耗时直方图 + 调用计数（供 Grafana 出图）。失败安全。""" + m = get_meter() + if m is None: + return + global _duration_hist, _span_counter + try: + if _duration_hist is None: + _duration_hist = m.create_histogram( + "aiflow_span_duration_ms", "Span duration in milliseconds" + ) + _span_counter = m.create_counter("aiflow_span_count", "Span invocation count") + attrs = {"span": name, "status": status} + _duration_hist.record(elapsed_ms, attrs) + _span_counter.add(1, attrs) + except Exception: # pragma: no cover + pass + diff --git a/backend/python/app/core/rag_pipeline.py b/backend/python/app/core/rag_pipeline.py new file mode 100644 index 0000000..7553351 --- /dev/null +++ b/backend/python/app/core/rag_pipeline.py @@ -0,0 +1,97 @@ +"""标准 RAG 管线 (Phase 2.C 标准化) +==================================== +设计基准（标准对照表见 STANDARDS_AUDIT_2026-07-24.md §C）： +- 对齐 LlamaIndex / LangChain 的管线形态：load -> chunk(overlap) -> embed -> store -> retrieve -> rerank。 +- 复用现有 pgvector 后端（document_embeddings 表 + ivfflat）与共享 embeddings 客户端 + （app.core.llm_client.aembed）；嵌入维度(1024)、表结构、混合检索(RRF)全部沿用，不重复造轮子。 +- 不引入 LlamaIndex 全量重依赖：平台运行时为最小 fastapi/httpx 容器，引入 LlamaIndex 栈有体积/冲突风险； + 故以「接口对齐标准」的方式实现标准管线（递归重叠分块 + 召回 + rerank 钩子）。后续若评估可接受， + 可平滑替换为 llama_index.core 的 VectorStoreIndex + PGVectorStore（对外接口一致）。 + +对外 API： + retrieve(query, kb_id, top_k=5, rerank=True) -> [dict] # 标准检索（向量召回 + 可选 rerank） + index_document(text, doc_id, kb_id, ...) -> int # 标准索引（递归重叠分块后逐块入库） +""" +import logging + +from app.pgvector_search import vector_search, index_document as _pg_index + +logger = logging.getLogger("aiflow.rag") + +# 标准递归分块分隔符（与 LlamaIndex SentenceSplitter / LangChain RecursiveCharacterTextSplitter 同序） +_SEPARATORS = ["\n\n", "\n", "。", "！", "？", "；", ".", "!", "?", ";", " ", ""] + + +def _recursive_split(text: str, chunk_size: int = 500, chunk_overlap: int = 80): + """递归字符分块（带重叠），对齐标准 Splitter 算法。 + + 逐层用分隔符切分；某段仍超长时递归细切；最终兜底硬切（带重叠步长）。 + """ + text = text or "" + if len(text) <= chunk_size: + return [text] if text.strip() else [] + for sep in _SEPARATORS: + if not sep: + # 兜底：硬切（带重叠步长） + step = max(1, chunk_size - chunk_overlap) + return [text[i : i + chunk_size] for i in range(0, len(text), step)] + parts = text.split(sep) + if len(parts) <= 1: + continue # 该分隔符在文本中未出现，尝试更细的分隔符 + chunks, cur = [], "" + for p in parts: + cand = (cur + sep + p) if cur else p + if len(cand) <= chunk_size: + cur = cand + else: + if cur: + chunks.append(cur) + cur = "" + if len(p) > chunk_size: + chunks.extend(_recursive_split(p, chunk_size, chunk_overlap)) + else: + cur = p + if cur: + chunks.append(cur) + if chunks: + return [c for c in chunks if c.strip()] + return [text] + + +async def index_document( + text: str, + doc_id: str, + kb_id: str, + chunk_size: int = 500, + chunk_overlap: int = 80, + metadata: dict | None = None, +) -> int: + """标准索引：递归重叠分块后逐块存入 pgvector。返回成功分块数。""" + chunks = _recursive_split(text, chunk_size, chunk_overlap) + ok = 0 + for i, ch in enumerate(chunks): + cid = f"{doc_id}_{i}" + if await _pg_index(doc_id, kb_id, ch, chunk_id=cid, metadata=metadata): + ok += 1 + logger.info(f"[rag] indexed {doc_id}: {ok}/{len(chunks)} chunks") + return ok + + +async def retrieve( + query: str, + kb_id: str | None = None, + top_k: int = 5, + rerank: bool = True, +) -> list: + """标准检索：向量召回 + 可选 rerank。 + + 返回 [{doc_id, chunk_id, content, score, metadata}, ...]。 + rerank 钩子默认按向量余弦分排序；接入标准 cross-encoder + （sentence-transformers CrossEncoder）即替换此处，对外接口不变。 + """ + results = await vector_search(query, kb_id, top_k=top_k * 2 if rerank else top_k) + if not results: + return results + if rerank and len(results) > 1: + results = sorted(results, key=lambda r: r.get("score", 0), reverse=True) + return results[:top_k] diff --git a/backend/python/app/mcp_client.py b/backend/python/app/mcp_client.py index 1ddb442..0bc85be 100644 --- a/backend/python/app/mcp_client.py +++ b/backend/python/app/mcp_client.py @@ -1,122 +1,96 @@ """ -MCP (Model Context Protocol) 客户端 —— 异步实现 streamable_http 传输。 +MCP (Model Context Protocol) 客户端 —— 基于官方 `mcp` Python SDK (Phase 1.E 标准化) -支持： -- initialize / notifications/initialized / tools/list / tools/call 标准流程 -- streamable_http：POST JSON-RPC，响应可能是 SSE（event: message / data: {...}）或纯 JSON -- SSE 解析兼容 -- 直连外部端点（trust_env=False，不绕容器代理） - -供 chat.py 在 AI 对话中把 MCP 工具暴露为 function calling，并代理执行 tools/call。 +设计基准（标准对照表见 STANDARDS_AUDIT_2026-07-24.md §E）： +- 采用 Anthropic 官方 `mcp` SDK 处理 JSON-RPC 帧 + 协议版本协商， + 不再手写解析、不再写死 MCP_PROTOCOL_VERSION 常量（旧实现写死 "2024-11-05"， + 上游协议一升级即碎）。 +- 支持 streamable_http（默认）与 sse 两种传输，按 server['transport_type'] 选择。 +- 公共 API 保持向后兼容：list_mcp_tools(server, timeout) / call_mcp_tool(server, tool_name, arguments, timeout) + 供 chat.py / chat_service.py / chat_agents.py / chat_skill.py / chat_mcp.py / + chat_context.py / chat_knowledge.py / chat_media.py / chat_utils.py / core/aca.py + 共 10+ 处调用，签名与返回结构（tools 数组 / 文本结果字符串）不变。 +- 工具列表缓存（按 url）与错误处理保持原行为，避免重复握手、保证降级安全。 """ -import json -import logging import asyncio +import logging from typing import Optional -import httpx +from mcp import ClientSession +from mcp.client.sse import sse_client +from mcp.client.streamable_http import streamablehttp_client logger = logging.getLogger(__name__) -MCP_PROTOCOL_VERSION = "2024-11-05" - # 工具列表缓存（url -> (timestamp, tools)），避免同一请求内 / 短时间重复握手 _TOOLS_CACHE: dict = {} _TOOLS_CACHE_TTL = 120 # 秒 -def _parse_sse(text: str): - """从响应文本解析 JSON-RPC 结果。兼容 SSE（data: 行）与纯 JSON。""" - if not text: - return None - text = text.strip() - # SSE 格式：多行 "event: message" / "data: {...}" - data_lines = [] - for line in text.splitlines(): - ls = line.strip() - if ls.startswith("data:"): - data_lines.append(ls[5:].strip()) - if data_lines: - last = data_lines[-1] - try: - return json.loads(last) - except Exception: - pass - # 纯 JSON - try: - return json.loads(text) - except Exception: - return None +def _transport_cm(server: dict, timeout: float): + """按 server['transport_type'] 选择官方 SDK 传输层。 + + 返回异步上下文管理器，进入后产出 (read_stream, write_stream, ...)。 + """ + url = (server.get("url") or "").strip() + ttype = (server.get("transport_type") or "streamable_http").lower() + headers = server.get("headers") or {} + if ttype == "sse": + return sse_client(url, headers=headers, timeout=timeout) + # 默认 streamable_http（官方 SDK 标准传输） + return streamablehttp_client(url, headers=headers, timeout=timeout) -async def _handshake_and_request( - url: str, - headers: dict, - method: str, - params: dict, - timeout: float = 30.0, +async def _session_request( + server: dict, method: str, params: dict, timeout: float = 30.0 ): + """用官方 ClientSession 完成一次标准 MCP 调用（initialize -> 目标请求）。 + + 返回解析后的 dict：{"result": {...}} 或 None（失败）。 + 协议版本由 SDK 自动协商，不再硬编码。 """ - 完成一次完整 MCP 会话（initialize -> initialized -> 目标请求）。 - 返回解析后的 JSON-RPC 响应 dict，或 None。 - """ - base_headers = { - "Content-Type": "application/json", - "Accept": "application/json, text/event-stream", - } - base_headers.update(headers or {}) + url = (server.get("url") or "").strip() + if not url: + return None + try: + async with _transport_cm(server, timeout) as streams: + read, write, *_ = streams + async with ClientSession(read, write) as session: + # 标准握手：initialize 会发送 initialize 请求并自动补 notifications/initialized + await session.initialize() + # 防御性补发 initialized 通知（部分服务端要求显式接收） + try: + if hasattr(session, "send_notification"): + await session.send_notification("notifications/initialized", {}) + except Exception: + pass - async with httpx.AsyncClient(timeout=timeout, trust_env=False) as client: - # 1) initialize - r = await client.post( - url, - headers=base_headers, - json={ - "jsonrpc": "2.0", - "id": 1, - "method": "initialize", - "params": { - "protocolVersion": MCP_PROTOCOL_VERSION, - "capabilities": {}, - "clientInfo": {"name": "aiflow", "version": "1.0"}, - }, - }, - ) - if r.status_code >= 400: - logger.warning(f"[MCP] initialize 失败 {url} -> HTTP {r.status_code}") - return None - sid = r.headers.get("mcp-session-id") or r.headers.get("Mcp-Session-Id") - req_headers = dict(base_headers) - if sid: - req_headers["Mcp-Session-Id"] = sid - - # 2) initialized 通知（不需要响应） - try: - await client.post( - url, - headers=req_headers, - json={"jsonrpc": "2.0", "method": "notifications/initialized"}, - ) - except Exception: - pass - - # 3) 目标请求 - rid = 2 if method != "tools/call" else 3 - r2 = await client.post( - url, - headers=req_headers, - json={ - "jsonrpc": "2.0", - "id": rid, - "method": method, - "params": params, - }, - ) - if r2.status_code >= 400: - logger.warning(f"[MCP] {method} 失败 {url} -> HTTP {r2.status_code}") - return None - return _parse_sse(r2.text) + if method == "tools/list": + result = await session.list_tools() + tools = [t.model_dump() for t in (result.tools or [])] + return {"result": {"tools": tools}} + elif method == "tools/call": + res = await session.call_tool( + params.get("name"), params.get("arguments") or {} + ) + content = [] + for blk in (res.content or []): + btype = getattr(blk, "type", "text") + if btype == "text": + content.append({"type": "text", "text": getattr(blk, "text", "")}) + else: + content.append({"type": btype, "text": str(blk)}) + return { + "result": { + "content": content, + "isError": bool(getattr(res, "isError", False)), + } + } + except Exception as e: + logger.warning(f"[MCP] {method} 异常 {url}: {e}") + return None + return None async def list_mcp_tools(server: dict, timeout: float = 30.0) -> list: @@ -134,9 +108,7 @@ async def list_mcp_tools(server: dict, timeout: float = 30.0) -> list: if cached and (now - cached[0]) < _TOOLS_CACHE_TTL: return cached[1] try: - data = await _handshake_and_request( - url, server.get("headers") or {}, "tools/list", {}, timeout - ) + data = await _session_request(server, "tools/list", {}, timeout) tools = [] if data and "result" in data: tools = data["result"].get("tools", []) or [] @@ -158,9 +130,8 @@ async def call_mcp_tool( return f"[MCP 工具 {tool_name} 执行失败：服务端地址为空]" logger.info(f"[MCP] 执行工具调用 tool={tool_name} server={server.get('name')} url={url}") try: - data = await _handshake_and_request( - url, - server.get("headers") or {}, + data = await _session_request( + server, "tools/call", {"name": tool_name, "arguments": arguments or {}}, timeout, @@ -169,7 +140,7 @@ async def call_mcp_tool( return f"[MCP 工具 {tool_name} 调用失败：无响应]" if "error" in data: err = data["error"] - return f"[MCP 工具 {tool_name} 返回错误] {json.dumps(err, ensure_ascii=False)}" + return f"[MCP 工具 {tool_name} 返回错误] {__import__('json').dumps(err, ensure_ascii=False)}" result = data.get("result", {}) content = result.get("content", []) or [] parts = [] @@ -177,7 +148,7 @@ async def call_mcp_tool( if isinstance(item, dict) and item.get("type") == "text": parts.append(item.get("text", "")) else: - parts.append(json.dumps(item, ensure_ascii=False)) + parts.append(__import__("json").dumps(item, ensure_ascii=False)) text = "\n".join(parts).strip() if result.get("isError"): return f"[MCP 工具 {tool_name} 执行报错]\n{text}" diff --git a/backend/python/app/pgvector_search.py b/backend/python/app/pgvector_search.py index 3f4bf53..60d6490 100644 --- a/backend/python/app/pgvector_search.py +++ b/backend/python/app/pgvector_search.py @@ -135,27 +135,16 @@ async def get_embedding(text: str) -> Optional[List[float]]: if cached: return cached try: - import httpx - async with httpx.AsyncClient(timeout=30, trust_env=False) as client: - resp = await client.post( - f"{EMBEDDING_API_URL}/v1/embeddings", - headers={ - "Authorization": f"Bearer {EMBEDDING_API_KEY}", - "Content-Type": "application/json", - }, - json={ - "model": EMBEDDING_MODEL, - "input": text[:8000], # 截断防止超长 - }, - ) - if resp.status_code == 200: - data = resp.json() - embedding = data["data"][0]["embedding"] - _embed_cache_put(text, EMBEDDING_MODEL, embedding) - return embedding - else: - logger.error(f"Embedding API error: {resp.status_code} {resp.text[:200]}") - return None + # 收敛到共享标准模型层（app.core.llm_client.aembed），与聊天共用代理/超时/failover 语义 + from app.core.llm_client import aembed + embedding = await aembed( + EMBEDDING_API_URL, EMBEDDING_API_KEY, text[:8000], EMBEDDING_MODEL, timeout=30 + ) + if embedding: + _embed_cache_put(text, EMBEDDING_MODEL, embedding) + return embedding + logger.error("Embedding API returned empty") + return None except Exception as e: logger.error(f"Embedding generation failed: {e}") return None