qyyfw/qihuaiflow
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
a7a57f9a03607e21c3c14ac3f2f9b924a15e5173
qihuaiflow/backend/python/app/api/chat_service.py
T

2046 lines
29 KiB
Python
Raw Blame History

# 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
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)
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 = "<<CLARIFY>>" + json.dumps(
{"prompt": req.content, "options": clarify["options"]}, ensure_ascii=False
) + "<<CLARIFY>>"
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)
if external_agent:
agents = external_agents
# 组装最近历史(供 n8n 等无平台会话态的服务维持多轮上下文)
_hist_rows = db.query(Message).filter(
Message.conversation_id == conv_id
).order_by(asc(Message.created_at)).limit(10).all()
_history = [{"role": h.role, "content": h.content}
for h in _hist_rows if h.role in ("user", "assistant")]
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}]
# 加载对话历史(最多10条)
history = db.query(Message).filter(
Message.conversation_id == conv_id
).order_by(asc(Message.created_at)).limit(10).all()
for h in history:
if h.role == "user" or h.role == "assistant":
messages.append({"role": h.role, "content": h.content})
# Issue 1: 必须把附件内容(attachment_block)一并发给模型，
# 否则用户在对话窗上传文件后模型“看不到”文件内容。
messages.append({"role": "user", "content": req.content + attachment_block})
# Stage B.3: Function Calling循环(最多5轮)
max_tool_rounds = 5
for round_num in range(max_tool_rounds):
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", "")
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 "<img" in _ingest_text
or "![]( " in _ingest_text
):
try:
asyncio.create_task(_ingest_answer_to_dify(conv_id, ai_msg.id, _ingest_text))
except Exception:
pass
# 更新used_skills / used_mcp：优先反馈用户显式指定的；否则反馈本轮实际调用的
used_skills = []
used_mcp = []
if effective_skill_ids:
# 用户显式选择了技能 -> 反馈配对的技能名
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,
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()
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": "发送成功",
}
Reference in new issue View Git Blame Copy Permalink