From c32eed39bdee1bd787cd212706206c23840abedf Mon Sep 17 00:00:00 2001 From: qihu-dev Date: Wed, 22 Jul 2026 00:17:54 +0800 Subject: [PATCH] =?UTF-8?q?feat(chat):=20=E6=94=B6=E6=8B=A2=E5=BC=8FAI?= =?UTF-8?q?=E6=80=9D=E8=80=83/=E6=89=A7=E8=A1=8C=E8=BF=87=E7=A8=8B?= =?UTF-8?q?=E5=8F=AF=E8=A7=86=E5=8C=96=20+=20=E4=BF=AE=E5=A4=8Dspan(name)?= =?UTF-8?q?=E9=87=8D=E5=A4=8D=E4=BC=A0=E5=8F=82=E5=AF=BC=E8=87=B4=E6=8A=80?= =?UTF-8?q?=E8=83=BD/MCP=E6=89=A7=E8=A1=8C=E5=B4=A9=E6=BA=83?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/python/app/api/chat_service.py | 6396 ++++++++++++++++-------- frontend/src/pages/front/ChatPage.css | 107 + frontend/src/pages/front/ChatPage.tsx | 30 + frontend/src/types/index.ts | 7 + 4 files changed, 4417 insertions(+), 2123 deletions(-) diff --git a/backend/python/app/api/chat_service.py b/backend/python/app/api/chat_service.py index bda1f8b..e24614b 100644 --- a/backend/python/app/api/chat_service.py +++ b/backend/python/app/api/chat_service.py @@ -1,2123 +1,4273 @@ -# 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 - - - - 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) - - - - - - - - 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 = _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), - - ) - - - - - - - - # 构建消息列表 - - - - # 实际调用追踪（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)一并发给模型， - - - - # 否则用户在对话窗上传文件后模型“看不到”文件内容。 - - - - 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", []) - - - - 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("mcp.call", name=_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", "")) - - else: - - async with span("skill.execute", name=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", "")) - - - - 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 模式下将自动使用平台默认模型。" - - - - - - - - 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 - - - - - - - - # 更新对话统计 - - - - conv.knowledge_graph_count = (conv.knowledge_graph_count or 0) + 1 - - - - 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}") - - - - - - - - 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, - - - - }, - - - - }, - - - - "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) + + + +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) + + + + + + + + + + + + + + + + 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), + + + + ) + + + + + + + + + + + + + + + + # 构建消息列表 + + + + + + + + # 实际调用追踪（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)一并发给模型， + + + + + + + + # 否则用户在对话窗上传文件后模型“看不到”文件内容。 + + + + + + + + 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 模式下将自动使用平台默认模型。" + + + + + + + + + + + + + + + + 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 + + + + + + + + + + + + + + + + # 更新对话统计 + + + + + + + + conv.knowledge_graph_count = (conv.knowledge_graph_count or 0) + 1 + + + + + + + + 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}") + + + + + + + + + + + + + + + + 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/frontend/src/pages/front/ChatPage.css b/frontend/src/pages/front/ChatPage.css index afd0eb5..8b3ac58 100644 --- a/frontend/src/pages/front/ChatPage.css +++ b/frontend/src/pages/front/ChatPage.css @@ -1557,3 +1557,110 @@ del { border-color: var(--color-accent, #5B5FED); color: var(--color-accent, #5B5FED); } + +/* ===== AI 思考 / 执行过程面板（收拢式） ===== */ +.ai-process { + margin-top: 10px; + border: 1px solid var(--color-border-light, #e6e9f0); + border-radius: var(--radius-md, 12px); + background: color-mix(in srgb, var(--color-bg-secondary, #fff) 75%, var(--color-accent, #5B5FED) 4%); + backdrop-filter: blur(8px) saturate(160%); + overflow: hidden; + transition: border-color var(--transition-fast, .2s), box-shadow var(--transition-fast, .2s); +} + +.ai-process[open] { + box-shadow: 0 6px 22px rgba(91, 95, 237, 0.10); + border-color: color-mix(in srgb, var(--color-accent, #5B5FED) 35%, transparent); +} + +.ai-process-summary { + list-style: none; + cursor: pointer; + display: flex; + align-items: center; + gap: 8px; + padding: 9px 14px; + font-size: var(--text-sm, 13px); + font-weight: 600; + color: var(--color-text-secondary, #475069); + user-select: none; + transition: background var(--transition-fast, .2s); +} + +.ai-process-summary::-webkit-details-marker { display: none; } +.ai-process-summary::marker { content: ""; } + +.ai-process-summary:hover { + background: color-mix(in srgb, var(--color-accent, #5B5FED) 7%, transparent); +} + +.ai-process-badge { + font-size: 15px; + line-height: 1; + filter: drop-shadow(0 1px 2px rgba(91, 95, 237, 0.25)); +} + +.ai-process-list { + padding: 4px 14px 12px; + display: flex; + flex-direction: column; + gap: 8px; + animation: ai-process-in .28s cubic-bezier(0.16, 1, 0.3, 1); +} + +@keyframes ai-process-in { + from { opacity: 0; transform: translateY(-4px); } + to { opacity: 1; transform: translateY(0); } +} + +.ai-process-item { + display: flex; + gap: 10px; + padding: 9px 12px; + border-radius: 10px; + background: var(--color-bg-secondary, #fff); + border: 1px solid var(--color-border-light, #e6e9f0); + border-left: 3px solid var(--color-accent, #5B5FED); +} + +.ai-process-thinking { border-left-color: #8b8fe6; } +.ai-process-skill { border-left-color: #5B5FED; } +.ai-process-mcp { border-left-color: #12a594; } + +.ai-process-ico { + font-size: 15px; + line-height: 1.4; + flex-shrink: 0; +} + +.ai-process-body { + min-width: 0; + flex: 1; +} + +.ai-process-title { + display: block; + font-size: var(--text-sm, 13px); + font-weight: 600; + color: var(--color-text-primary, #1f2430); + margin-bottom: 2px; +} + +.ai-process-text { + margin: 0; + font-size: var(--text-xs, 12px); + line-height: 1.55; + color: var(--color-text-tertiary, #8a90a6); + max-height: 160px; + overflow-y: auto; + word-break: break-word; + white-space: pre-wrap; +} + +/* 暗色主题微调 */ +[data-theme="dark"] .ai-process, +html.dark .ai-process { + background: color-mix(in srgb, var(--color-bg-secondary, #1b1d29) 70%, var(--color-accent, #5B5FED) 8%); + border-color: color-mix(in srgb, var(--color-accent, #5B5FED) 25%, transparent); +} diff --git a/frontend/src/pages/front/ChatPage.tsx b/frontend/src/pages/front/ChatPage.tsx index bacbd3b..b38d86b 100644 --- a/frontend/src/pages/front/ChatPage.tsx +++ b/frontend/src/pages/front/ChatPage.tsx @@ -895,6 +895,7 @@ const ChatPage: React.FC = () => { createdAt: data.assistant_message.created_at || new Date().toISOString(), used_skills: data.assistant_message.used_skills || [], used_mcp: data.assistant_message.used_mcp || [], + process_steps: data.assistant_message.process_steps || [], }); } return [...filtered, ...newMsgs]; @@ -1168,6 +1169,7 @@ const ChatPage: React.FC = () => { // Real used_skills from API response const usedSkills: string[] = (msg.role === 'assistant' && msg.used_skills) ? msg.used_skills : []; const usedMcp: string[] = (msg.role === 'assistant' && msg.used_mcp) ? msg.used_mcp : []; + const steps: any[] = (msg as any).process_steps || []; // 模糊创作意图确认卡片 const clarify = msg.role === 'assistant' ? parseClarify(msg.content) : null; return ( @@ -1235,6 +1237,34 @@ const ChatPage: React.FC = () => { )} )} + {msg.role === 'assistant' && steps.length > 0 && ( +
+ + 🧠 + 思考 / 执行过程（{steps.length} 步，点击展开） + +
+ {steps.map((s: any, i: number) => ( +
+ + {s.type === 'thinking' ? '💡' : s.type === 'skill' ? '🛠' : '🔌'} + +
+ {s.type === 'thinking' && ( + <>思考

{s.text}

+ )} + {s.type === 'skill' && ( + <>调用技能：{s.name}

{(s.result || '').slice(0, 400)}

+ )} + {s.type === 'mcp' && ( + <>MCP · {s.server}：{s.name}

{(s.result || '').slice(0, 400)}

+ )} +
+
+ ))} +
+
+ )}
{formatTime(msg.createdAt)} {msg.tokenUsed} tokens diff --git a/frontend/src/types/index.ts b/frontend/src/types/index.ts index 0a71eae..aac145e 100644 --- a/frontend/src/types/index.ts +++ b/frontend/src/types/index.ts @@ -62,6 +62,13 @@ export interface Message { createdAt: string; used_skills?: string[]; used_mcp?: string[]; + process_steps?: Array<{ + type: 'thinking' | 'skill' | 'mcp'; + text?: string; + name?: string; + server?: string; + result?: string; + }>; }