qyyfw/qihuaiflow
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
13578ba4c2dd3ee586a14b0b3a3f5b7d362c8c92
qihuaiflow/backend/python/app/api/chat.py
T
QihuAI Design d593d8dc42 fix: 显式选中技能/MCP 时优先按工具资料回答 + 媒体任务超时/日志 
- _build_system_prompt 末尾增加【最高优先级覆盖规则】：用户明确选择技能/MCP 并询问其介绍/功能/用法时，
  必须只基于该工具资料作答，禁止引用平台知识库中技能市场/MCP/平台能力的通用描述。
- _generate_media_task 增加启动/结束日志，对图像生成加 90s 超时、视频 300s 超时，避免异步任务挂死导致前端一直显示“生成中”。
2026-07-18 13:52:26 +08:00

10071 lines
122 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""
增强版 Chat API — Phase 2.5
包含: Skill ID统一(MarketApp映射), 真实Skill执行, Agent工作流注入,
多轮上下文保持, BM25 KB检索, Function Calling循环增强
"""
from fastapi import APIRouter, Depends, HTTPException, Query, Response
from sqlalchemy.orm import Session
from sqlalchemy import and_, desc, asc, func
from pydantic import BaseModel, Field
from typing import Optional
import uuid
from datetime import datetime, timezone, timedelta
import re
import json
import asyncio
import math
import httpx
import os
import logging
from ..database import get_db, SessionLocal
from ..models import (
User, Conversation, Message, AIModel, ModelProvider, TokenUsageLog,
KnowledgeBase, KnowledgeDocument, KnowledgeDocumentChunk, EvolutionSkill,
EvolutionTrajectory, MarketApp, AppPurchase, AppInstallation,
MCPServer,
gen_uuid
)
from ..auth import get_current_user
from ..platform_knowledge import get_platform_knowledge
from ..mcp_client import list_mcp_tools, call_mcp_tool
from ..middleware.log_service import LogService
from ..capability_registry import resolve_auto_model
router = APIRouter()
logger = logging.getLogger(__name__)
def detect_media_intent(text: str) -> str | None:
"""根据用户输入判断是否为 图像/视频/语音 生成请求，返回对应能力或 None。
用于 Auto 模式：用户未指定具体模型时，依据意图自动匹配平台各类能力的默认模型。
疑问句（含「吗/呢/怎么/能否」或结尾问号）不触发，避免「你能生成视频吗」被误判。
"""
if not text:
return None
t = text.lower().strip()
# 疑问句不触发媒体生成
if t.endswith(("？", "?")) or any(q in t for q in ("吗", "呢", "怎么", "如何", "能否", "是否可以", "可以吗", "会不会", "why", "how")):
return None
video_kw = ("视频", "短片", "影片", "vide", "动画视频", "一段视频", "做个视频", "生成视频",
"视频生成", "来段视频", "拍个视频", "做段视频", "视频剪辑", "做视频",
"生成一个视频", "制作一个视频", "帮我做视频", "制作视频",
"生成短片", "做短片", "制作短片", "一段短片")
image_kw = ("图片", "图像", "插画", "海报", "照片", "头像", "壁纸", "漫画", "表情包", "logo", "图标",
"imag", "画一张", "画张", "画一只", "画只", "画个", "画幅",
"来张", "来一张", "来幅", "发张图", "发个图",
"生成图片", "生成一张", "生成一幅", "生成图", "图像生成",
"做一张", "做张图", "做张", "出张图", "出一张图",
"帮我画", "给我画", "请画", "绘制", "作画", "ai绘画", "ai画图",
"画个图", "画张图", "画一幅图", "一张图", "一幅画", "生成一幅",
"画画", "画一下", "画出来", "生成个图", "生成一张图", "生成个图片",
"做图片", "做个图", "制图", "整张图", "整张图出来", "搞张图",
"帮我生成图片", "帮我生成图像", "给我生成图片", "给我生成图像",
"生成一张图片", "生成一个图片", "制作一张图", "制作一张图片",
"来一张图片", "画一张图片", "画一幅", "画一副")
audio_kw = ("语音", "配音", "朗读", "播报", "念一段", "有声", "tts", "音频", "生成语音",
"读一下", "念一下", "语音合成")
if any(k in t for k in video_kw):
return "video"
if any(k in t for k in image_kw):
return "image"
if any(k in t for k in audio_kw):
return "audio"
return None
def detect_ambiguous_creative_intent(text: str):
"""检测「模糊创作意图」：用户使用了创作类动词(制作/创作/做/生成/画/写/设计…)，
但未明确产出形式(文案/图片/视频/其他)。此时应与用户确认方向，而不是直接调用大模型
（某些模型会对这类开放式创作请求返回 "high risk" 拒答）。
返回 None 表示无需确认；否则返回 {"options": [...]}。
"""
if not text:
return None
t = text.strip()
tl = t.lower()
# 已有明确媒体意图(图像/视频/语音)则无需确认
if detect_media_intent(t):
return None
# 疑问句不触发
if tl.endswith(("？", "?")) or any(q in tl for q in ("吗", "呢", "怎么", "如何", "能否", "是否可以", "可以吗", "会不会", "why", "how", "什么", "哪些")):
return None
# 已明确给出产出形式(文字类/其他类)则无需确认
explicit_medium = (
"文案", "文章", "诗歌", "诗", "小说", "故事", "代码", "程序", "脚本", "总结", "翻译",
"邮件", "报告", "段落", "正文", "文字", "写作", "ppt", "演示", "思维导图", "表格",
"图表", "数据",
)
if any(m in tl for m in explicit_medium):
return None
# 创作类动词
creative_verbs = (
"制作", "创作", "做个", "做一个", "搞一个", "产出", "构思", "设想",
"打造", "塑造", "生成", "帮我做", "帮我生成", "帮我画", "帮我写", "请制作",
"请创作", "请生成", "请画", "请写", "来一个", "来一张", "来一幅", "做一张",
"做张", "写一首", "写一段", "写个",
)
if not any(v in t for v in creative_verbs):
return None
# 模糊：有创作动词但无明确产出形式 → 提供方向选择
options = [
{"key": "text", "label": "文案", "desc": "文章 / 故事 / 诗歌等文字内容", "medium": "文案"},
{"key": "image", "label": "图片", "desc": "插画 / 海报 / 照片等图像", "medium": "图片"},
{"key": "video", "label": "视频", "desc": "短片 / 动画等视频", "medium": "视频"},
{"key": "other", "label": "其他", "desc": "语音 / 代码 / 其他形式", "medium": "其他"},
]
return {"options": options}
# CST timezone (UTC+8) for consistent time display
CST_TZ = timezone(timedelta(hours=8))
def cst_now():
"""Get current time in CST timezone"""
return datetime.now(CST_TZ)
def _html_escape(s: str) -> str:
return (str(s).replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;")
.replace('"', "&quot;").replace("'", "&#39;"))
def _outbound_proxy() -> str:
"""出网代理地址(兜底)。默认经 172.18.0.1:7890 智能路由代理；
运行时若未注入代理环境变量则回退到该地址。"""
import os
return os.environ.get("HTTPS_PROXY") or os.environ.get("HTTP_PROXY") or "http://172.18.0.1:7890"
# 香港出网代理(实测对 AGNES 延迟更低、更稳定，2026-07-15 用户实测)。
# 含 Basic 鉴权，与服务器 /opt/proxy-router/proxy.py 的 UPSTREAM_USER/PASS 一致。
# 注意：智能路由代理 172.18.0.1:7890 对 AGNES 按 DOMESTIC_OVERRIDE 强制直连(曾返 502)，
# 故 AGNES 直接打 HK 代理，绕开智能路由代理。
_HK_PROXY = "http://qyyfw:XHR521yhf@103.242.14.208:11495"
def _proxy_for_url(url: str) -> str | None:
"""按目标域名选择是否走代理。
2026-07-15 调整：AGNES(apihub.agnes-ai.com / platform-outputs.agnes-ai.space)
经用户实测走香港代理 103.242.14.208:11495 延迟更低、更稳定，故改为走 HK 代理
(带鉴权)，不再直连，也不经智能路由代理(172.18.0.1:7890 对 AGNES 返 502)。
其余国内/平台域名直连，真正的境外域名走 _outbound_proxy()。
注意：httpx 默认 trust_env=True 会忽略 proxy 参数而改用环境变量 HTTP_PROXY，
故调用方必须同时传 trust_env=False 才能真正走此处指定的代理。
"""
import os
from urllib.parse import urlparse
host = (urlparse(url).hostname or "").lower()
if "agnes-ai.com" in host or "agnes-ai.space" in host:
return _HK_PROXY
# 平台自有域名(含自建 Dify/n8n 等)必须直连，不经出网代理。
if "qihuyun.com" in host:
return None
# 尊重 NO_PROXY：国内域名(硅基流动/DeepSeek/小米MoE 等)直连更稳定，
# 避免经易波动的智能路由代理 172.18.0.1:7890(对国内域名偶发连接重置)。
_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 _outbound_proxy()
# ============================================================
# 媒体中转(relay)端点：让微信/iLink 能内联渲染 AGNES 生成的图片/视频
# 背景：AGNES 媒体 URL 位于境外(agnes-ai.com / agnes-ai.space)，
# iLink 服务器(中国大陆)直连常被墙/超时，导致微信里只看到"链接"无法内联播放。
# 此端点由平台服务器(经香港代理可达 AGNES)代取媒体并以平台域名回传，
# 微信即可内联显示。仅允许白名单域名，杜绝 SSRF。
# ============================================================
_MEDIA_RELAY_ALLOWED = ("agnes-ai.com", "agnes-ai.space", "qihuyun.com")
@router.get("/media/relay")
async def media_relay(url: str = Query(...)):
"""中转 AGNES 媒体(图片/视频)到平台域名，供微信/iLink 内联渲染。
仅允许白名单域名；否则 400。返回的媒体流保持原 content-type。
"""
import urllib.parse as _up
host = (_up.urlparse(url).hostname or "").lower()
if not any(allowed in host for allowed in _MEDIA_RELAY_ALLOWED):
raise HTTPException(status_code=400, detail="unsupported media host")
try:
async with httpx.AsyncClient(timeout=180, proxy=_proxy_for_url(url), trust_env=False) as client:
r = await client.get(url)
if r.status_code != 200:
raise HTTPException(status_code=r.status_code, detail="upstream error")
content_type = r.headers.get("content-type", "application/octet-stream")
data = await r.aread()
return Response(content=data, media_type=content_type)
except HTTPException:
raise
except Exception as e:
raise HTTPException(status_code=502, detail=f"relay failed: {e}")
async def _optimize_media_prompt(raw_prompt: str, media_type: str, db=None) -> str:
"""使用平台文本模型对用户原始输入做「媒体提示词增强优化」。
AGNES 图像/视频模型对 prompt 质量高度敏感：
- 用户在多轮对话中说"帮我生成一张刚才讨论的产品图"包含大量会话噪音
- AGNES 需要的是**详细的、描述性的、英文优先或中英混合**的视觉提示词
- 缺乏构图/风格/光影/质量等指导词时，生成结果与预期差异巨大
本函数把用户原文交给文本 LLM 精炼为符合 AGNES 预期的优质 prompt。
失败时回退到原始输入(不做额外引导，保证必达)。
"""
if not raw_prompt or len(raw_prompt.strip()) < 2:
return raw_prompt
# 如果原始输入已经很长(>200 字)且英文为主，可能本身就是一个不错的 prompt，直接返回
stripped = raw_prompt.strip()
ascii_chars = sum(1 for c in stripped if ord(c) < 128)
if len(stripped) > 200 and ascii_chars / max(len(stripped), 1) > 0.6:
return stripped
# 用文本 LLM 做翻译增强
try:
from ..capability_registry import resolve_auto_model as _resolve_auto
from ..models import ModelProvider as MP
from ..database import SessionLocal as SL
import httpx as _httpx
# 独立数据库会话，避免与主流程竞争
_db = db
_should_close = False
if _db is None:
_db = SL()
_should_close = True
try:
text_model = _resolve_auto("text", _db)
if not text_model:
return stripped
provider = _db.query(MP).filter(MP.id == text_model.provider_id).first()
api_key = text_model.api_key or (provider.api_key if provider else None)
api_url = text_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 not (api_key and api_url):
return stripped
prompt_zh = (
"你是一个专业的AI图像/视频提示词工程师。请将以下用户需求提炼成"
f"一段高质量的{'图像' if media_type == 'image' else '视频'}生成"
"提示词。要求：\n"
"1. 用英文书写（AGNES模型对英文响应最佳），核心关键词可保留中文；\n"
"2. 添加构图/风格/光影/色彩/质量描述（如 highly detailed, 8K, "
"professional lighting, masterpiece, cinematic等）；\n"
"3. 去掉与视觉无关的对话噪音；\n"
"4. 只输出最终prompt文本，不要任何解释、不要引号包裹、不要\"Prompt:\"前缀。\n\n"
f"用户需求：{stripped}"
)
messages = [
{"role": "system", "content": "You are a professional prompt engineer. Output ONLY the final prompt text, no explanations."},
{"role": "user", "content": prompt_zh},
]
# max_tokens 调大到 1024：mimo 等推理模型会把 token 消耗在思考链(reasoning_content)
# 上，512 时 content 常被截断为空；并关闭推理(enable_thinking=False)，让模型直接输出提示词。
# 对不支持该参数的模型至多优化失败 → 下方 except 兜底退化为原始提示词，不阻断生成。
body = {"model": text_model.name, "messages": messages, "max_tokens": 1024,
"temperature": 0.7, "enable_thinking": False}
async with _httpx.AsyncClient(
timeout=30.0, proxy=_proxy_for_url(api_url), trust_env=False
) as client:
resp = await client.post(
api_url,
headers={"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"},
json=body,
)
if resp.status_code == 200:
data = resp.json()
opt = data.get("choices", [{}])[0].get("message", {}).get("content", "").strip()
if opt and len(opt) > 5:
return opt
finally:
if _should_close:
_db.close()
except Exception:
pass
return stripped
def _friendly_media_err(kind: str, raw: str) -> str:
"""把 AGNES 原始错误转译为对用户友好的中文。
AGNES 常见错误：429 限频(每分钟1次视频) / Server disconnected(连接重置) /
publish_video_queue_failed + context deadline exceeded(队列超时)。这些都不是
平台代码问题，统一给出可操作的提示，不把原始 HTTP/JSON 甩给用户。
"""
r = (raw or "").lower()
if "429" in raw or "rate limit" in r:
return (f"⚠️ {kind}生成请求过于频繁（AGNES 限制每分钟 1 次），"
f"请稍等 1 分钟后再试。")
if "server disconnected" in r or "连接" in raw or "reset" in r:
return f"⚠️ {kind}生成服务暂时连接失败，请稍后重试。"
if "publish_video" in r or "queue" in r or "deadline" in r or "context" in r:
return f"⚠️ {kind}生成队列繁忙（AGNES 处理超时），请稍后重试。"
return f"⚠️ {kind}生成失败（请稍后重试）：{(raw or '')[:120]}"
async def _agnes_post_json(api_url: str, headers: dict, payload: dict, timeout) -> tuple:
"""直连 AGNES 发 POST，对 5xx / 连接中断做指数退避重试。
AGNES 视频/图像队列极不稳定：经常返回 500 publish_video_queue_failed /
context deadline exceeded，或连接被重置(Server disconnected)。这类错误是 AGNES
侧瞬时抖动，重试通常立刻成功。4xx（参数错误）不重试，直接返回。
返回 (resp, err)：成功(resp.status_code<500) 时 err=None；耗尽重试时 resp=None。
"""
last_err = None
for attempt in range(3):
try:
async with httpx.AsyncClient(
timeout=timeout, proxy=_proxy_for_url(api_url), trust_env=False
) as client:
resp = await client.post(api_url, headers=headers, json=payload)
if resp.status_code < 500:
return resp, None
last_err = f"HTTP {resp.status_code}: {resp.text[:200]}"
except Exception as e:
last_err = str(e)[:200]
if attempt < 2:
await asyncio.sleep(2 * (attempt + 1))
return None, last_err
async def _generate_media(db, ai_model, prompt: str, user_id: str = None) -> str:
"""图像/视频生成模型统一入口：按 capabilities 调用 AGNES 等媒体生成 API。
返回可直接在 AI 聊天中渲染的富文本(含 <img>/<video> 标签)。"""
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)
if not api_key:
return "⚠️ 该媒体模型缺少 API Key，请在后台「大模型库」中配置。"
caps = [c.lower() for c in (ai_model.capabilities or [])]
headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
timeout = ai_model.timeout or 120
# ============ 图像生成 ============
if "image" in caps:
api_url = ai_model.endpoint
if not api_url:
api_url = f"{provider.api_base_url.rstrip('/')}/v1/images/generations" if provider else ""
if not api_url:
return "⚠️ 该图像模型未配置 Endpoint。"
# 先优化 prompt（用户原始输入含多轮对话噪音 → 高质量图像生成提示词）
# 优化失败绝不中断生成，退化为原始提示词
try:
optimized_prompt = await _optimize_media_prompt(prompt, "image", db)
except Exception:
optimized_prompt = prompt
payload = {
"model": ai_model.name,
"prompt": optimized_prompt,
"size": "1024x768",
}
resp, err = await _agnes_post_json(api_url, headers, payload, timeout)
if resp is None:
# 3 次重试后仍失败 → AGNES 侧持续抖动，提示稍后重试而非甩原始错误
return _friendly_media_err("图像", err)
if resp.status_code != 200:
# AGNES 内容策略拒绝：用原始(未优化)提示词重试一次，避免优化后提示词触发误杀
if resp.status_code == 400 and "content_policy" in (resp.text or "").lower():
payload["prompt"] = prompt
resp2, _ = await _agnes_post_json(api_url, headers, payload, timeout)
if resp2 is not None and resp2.status_code == 200:
resp = resp2
else:
return "⚠️ 图像生成被内容策略拒绝，请换一个更中性的描述后重试。"
else:
return _friendly_media_err("图像", f"HTTP {resp.status_code}: {resp.text[:200]}")
try:
data = resp.json()
except Exception:
return f"⚠️ 图像生成失败：响应解析错误 {resp.text[:200]}"
items = data.get("data") or []
if not items:
return f"⚠️ 图像生成失败：响应无图片数据 {str(data)[:200]}"
item = items[0]
url = item.get("url") or item.get("b64_json")
if not url:
return f"⚠️ 图像生成失败：未返回图片地址 {str(data)[:200]}"
if not url.startswith("http"):
url = f"data:image/png;base64,{url}"
return (f'<img class="md-img" src="{url}" alt="生成图片"/>\n\n'
f'**提示词：** {_html_escape(prompt)}')
# ============ 视频生成 (异步任务 + 轮询) ============
if "video" in caps:
api_url = ai_model.endpoint
if not api_url:
api_url = f"{provider.api_base_url.rstrip('/')}/v1/videos" if provider else ""
if not api_url:
return "⚠️ 该视频模型未配置 Endpoint。"
# 优化 prompt（用户输入含多轮对话噪音 → 高质量视频提示词）；失败退化为原始提示词
try:
optimized_prompt = await _optimize_media_prompt(prompt, "video", db)
except Exception:
optimized_prompt = prompt
base = api_url.split("/v1")[0] if "/v1" in api_url else api_url
payload = {
"model": ai_model.name,
"prompt": optimized_prompt,
"height": 768,
"width": 1152,
"num_frames": 121,
"frame_rate": 24,
}
resp, err = await _agnes_post_json(api_url, headers, payload, timeout)
if resp is None:
return _friendly_media_err("视频", err)
if resp.status_code != 200:
# AGNES 内容策略拒绝：用原始(未优化)提示词重试一次，避免优化后提示词触发误杀
if resp.status_code == 400 and "content_policy" in (resp.text or "").lower():
payload["prompt"] = prompt
resp2, _ = await _agnes_post_json(api_url, headers, payload, timeout)
if resp2 is not None and resp2.status_code == 200:
resp = resp2
else:
return "⚠️ 视频生成被内容策略拒绝，请换一个更中性的描述后重试。"
else:
return _friendly_media_err("视频", f"HTTP {resp.status_code}: {resp.text[:200]}")
try:
data = resp.json()
except Exception:
return f"⚠️ 视频生成失败：响应解析错误 {resp.text[:200]}"
video_id = data.get("video_id") or data.get("task_id") or data.get("id")
if not video_id:
return f"⚠️ 视频生成失败：未返回任务ID {str(data)[:200]}"
# 轮询结果(最多约 5 分钟)。AGNES 视频状态端点为 GET {base}/v1/videos/{video_id}
# 该端点顶层返回 status(queued/in_progress/completed/failed)，成片地址在 metadata.url
result_url = None
err_msg = None
for _ in range(60):
await asyncio.sleep(5)
try:
async with httpx.AsyncClient(
timeout=timeout, proxy=_proxy_for_url(f"{base}/v1/videos/{video_id}"), trust_env=False
) as client:
r2 = await client.get(f"{base}/v1/videos/{video_id}", headers=headers)
except Exception:
continue
if r2.status_code != 200:
continue
rd = r2.json()
st = rd.get("status")
if st == "completed":
result_url = (rd.get("metadata") or {}).get("url") or rd.get("url")
break
if st == "failed":
err_msg = rd.get("error") or (rd.get("metadata") or {}).get("error")
break
if result_url:
return (f'<video class="md-video" controls src="{result_url}"></video>\n\n'
f'**提示词：** {_html_escape(prompt)}')
return f"⚠️ 视频生成失败：{err_msg or '生成超时，请稍后在对话中重试'}"
return "⚠️ 未知的媒体模型类型。"
async def _generate_media_task(ai_msg_id: str, ai_model_id: str, prompt: str, user_id: str):
"""后台异步执行图像/视频生成，完成后回写消息内容。
目的是把最长 5 分钟的 AGNES 视频轮询从 HTTP 请求中剥离，避免请求被客户端/
代理超时断开(表现为前端「发送失败：请求失败」)。send_message 立即返回占位
内容(generating 标记)，本任务独立在事件循环中轮询并在完成后 merge 回写消息。
使用独立的 SessionLocal 会话，因为请求级 db 在响应返回后即被 FastAPI 关闭。
"""
db = SessionLocal()
try:
ai_model = db.query(AIModel).filter(AIModel.id == ai_model_id).first()
if not ai_model:
logger.warning(f"[_generate_media_task] 模型不存在: {ai_model_id}")
return
logger.info(f"[_generate_media_task] 启动 media 生成 msg={ai_msg_id} model={ai_model_id} prompt={prompt[:60]!r}")
caps = [c.lower() for c in (ai_model.capabilities or [])]
is_video = "video" in caps
timeout_seconds = 300 if is_video else 90
try:
result = await asyncio.wait_for(
_generate_media(db, ai_model, prompt, user_id),
timeout=timeout_seconds
)
except asyncio.TimeoutError:
logger.warning(f"[_generate_media_task] media 生成超时 msg={ai_msg_id} is_video={is_video} timeout={timeout_seconds}s")
_media_name = "视频" if is_video else "图像"
result = f"⚠️ 媒体生成超时：{_media_name}生成耗时超过 {timeout_seconds} 秒，请稍后重试。"
except Exception as e:
logger.warning(f"[_generate_media_task] media 生成异常 msg={ai_msg_id}: {e}", exc_info=True)
result = f"⚠️ 媒体生成失败：{str(e)[:200]}"
# 回写消息内容(占位 → 最终富文本/错误)
msg = db.query(Message).filter(Message.id == ai_msg_id).first()
if msg:
msg.content = result
msg.token_used = len(result) // 2
db.commit()
logger.info(f"[_generate_media_task] 消息 {ai_msg_id} 已回写内容")
# 媒体 Token 记账：把图片/视频真实消耗写入 TokenUsageLog，供驾驶舱统计。
# 主流程对媒体请求只写了占位文本的几 token，此处补写真实媒体消耗（模型名正确、数值合理）。
_write_media_token_log(db, ai_model, user_id, msg.conversation_id if msg else None, prompt)
except Exception as e:
logger.warning(f"[_generate_media_task] 后台生成异常: {e}")
finally:
try:
db.close()
except Exception:
pass
def _write_media_token_log(db, ai_model, user_id: str, conversation_id: str, prompt: str):
"""媒体生成完成后补写 TokenUsageLog（图片/视频），修正此前只记占位文本的偏差。
媒体 API 不返回 token 用量，采用与 IM Bot 路径一致的估算：
视频基线 1200、图片基线 400，再加 prompt 长度的 1/4 作为提示词开销。
"""
if not user_id or not ai_model:
return
try:
is_video = "video" in (ai_model.capabilities or [])
base = 1200 if is_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,
conversation_id=conversation_id,
prompt_tokens=est,
completion_tokens=0,
total_tokens=est,
created_at=cst_now(),
))
db.commit()
except Exception as e:
logger.warning(f"[_write_media_token_log] 写入失败(忽略): {e}")
try:
db.rollback()
except Exception:
pass
# ============================================================
# 外部 Agent 远程调用 (Dify / n8n / Coze)
# ============================================================
# 当前台对话关联了一个带有效 agent_endpoint 的外部 Agent 时，运行期直接远程调用
# 该 Agent 平台，而不是用本地 LLM「扮演」。所有出网请求必须 trust_env=False +
# _proxy_for_url() 直连，避免容器 env 里的坏代理(172.18.0.1:7890 返回 502)。
def _normalize_dify_chat_url(endpoint: str) -> str:
"""把用户填写的 Dify 地址归一化为 /v1/chat-messages 端点。与 market.py 保持一致。"""
ep = (endpoint or "").rstrip("/")
if ep.endswith("/chat-messages"):
return ep
if ep.endswith("/v1"):
return ep + "/chat-messages"
return ep + "/v1/chat-messages"
def _extract_text_from_json(obj) -> str:
"""从 n8n / 通用 webhook 的任意 JSON 响应中尽力抽取文本回答。"""
if obj is None:
return ""
if isinstance(obj, str):
return obj
if isinstance(obj, (int, float, bool)):
return str(obj)
if isinstance(obj, list):
# 取第一个可抽取到文本的元素
for it in obj:
t = _extract_text_from_json(it)
if t:
return t
return ""
if isinstance(obj, dict):
# 常见回答字段优先级
for k in ("output", "text", "answer", "response", "reply",
"message", "content", "result", "data"):
if k in obj:
t = _extract_text_from_json(obj[k])
if t:
return t
# 兜底：整体序列化
try:
return json.dumps(obj, ensure_ascii=False)
except Exception:
return str(obj)
return str(obj)
async def _call_dify_agent(endpoint: str, api_key: str, query: str,
user_id: str, ext_conv_id: str | None) -> tuple:
"""调用 Dify(新版 Agent) /v1/chat-messages，流式(SSE)解析最终回答。
新版 Agent API 仅支持 response_mode=streaming(blocking 会 400)。事件流中
message / agent_message 事件携带增量 answer，累加即为最终回答；每个事件都带
conversation_id，用于维持多轮上下文。返回 (answer, new_conversation_id)。
"""
url = _normalize_dify_chat_url(endpoint)
headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
payload = {
"inputs": {},
"query": query,
"user": (user_id or "web-user")[:64],
"response_mode": "streaming",
"conversation_id": ext_conv_id or "",
}
answer_parts = []
new_conv_id = ext_conv_id
err_msg = None
wf_outputs = None
# httpx 默认 follow_redirects=False；stream POST 若遇 301/302 会直接返回该状态码。
# 且 httpx 自动跟随会把 POST 降级为 GET(对 /v1/chat-messages 变 405)。
# 因此手动跟随 301/302/307/308，保持 POST 与请求体不变(最多 4 跳)。
async with httpx.AsyncClient(timeout=180, proxy=_proxy_for_url(url), trust_env=False,
follow_redirects=False) as client:
cur_url = url
for _redir in range(4):
async with client.stream("POST", cur_url, headers=headers, json=payload) as resp:
if resp.status_code in (301, 302, 307, 308):
loc = resp.headers.get("location") or resp.headers.get("Location")
await resp.aclose()
if not loc:
raise RuntimeError(f"Dify HTTP {resp.status_code}: 缺少重定向地址")
# 处理相对地址
if loc.startswith("/"):
from urllib.parse import urlsplit, urlunsplit
p = urlsplit(cur_url)
loc = urlunsplit((p.scheme, p.netloc, loc, "", ""))
cur_url = loc
continue
if resp.status_code != 200:
body = (await resp.aread()).decode("utf-8", "replace")
raise RuntimeError(f"Dify HTTP {resp.status_code}: {body[:300]}")
async for line in resp.aiter_lines():
if not line:
continue
line = line.strip()
if not line.startswith("data:"):
continue
data_str = line[5:].strip()
if not data_str or data_str == "[DONE]":
continue
try:
evt = json.loads(data_str)
except Exception:
continue
ev = evt.get("event")
if evt.get("conversation_id"):
new_conv_id = evt["conversation_id"]
data = evt.get("data") or {}
# 兼容三类 Dify 应用：
# 1) Chat/Agent 应用: message / agent_message 事件带增量 answer
# 2) Chatflow/Workflow 应用: text_chunk 事件带 data.text；
# 结束时 workflow_finished 带 status / outputs / error
if ev in ("message", "agent_message"):
answer_parts.append(evt.get("answer", "") or "")
elif ev == "message_replace":
# 内容审查替换：以替换后的完整 answer 为准
answer_parts = [evt.get("answer", "") or ""]
elif ev == "text_chunk":
answer_parts.append((data.get("text") or ""))
elif ev == "node_finished":
# answer / end 节点的 outputs 可能含最终文本
if data.get("node_type") in ("answer", "end") and data.get("outputs"):
wf_outputs = data.get("outputs")
if data.get("status") == "failed" and data.get("error"):
err_msg = str(data.get("error"))
elif ev == "workflow_finished":
if data.get("status") == "failed":
err_msg = str(data.get("error") or "Dify 工作流执行失败")
elif data.get("outputs"):
wf_outputs = data.get("outputs")
elif ev == "error":
err_msg = evt.get("message") or evt.get("code") or "Dify 流式返回错误"
break
else:
raise RuntimeError("Dify 重定向次数过多(>4)，请直接填写重定向后的地址")
answer = "".join(answer_parts).strip()
# 流式无增量文本时，尝试从 workflow 结束 outputs 兜底提取
if not answer and wf_outputs:
answer = _extract_text_from_json(wf_outputs).strip()
if not answer and err_msg:
raise RuntimeError(err_msg)
return answer, new_conv_id
async def _call_n8n_agent(endpoint: str, api_key: str, query: str,
user_id: str, history: list) -> tuple:
"""调用 n8n 生产 Webhook (POST)。webhook 用 Respond to Webhook 节点返回 JSON，
尽力从中抽取文本回答。n8n webhook 无平台级会话态，多轮靠我们下发 history。
返回 (answer, None)。"""
url = (endpoint or "").strip()
headers = {"Content-Type": "application/json"}
if api_key:
# n8n Header Auth 自定义头名不固定，同时下发常见几种以提高兼容性
headers["Authorization"] = f"Bearer {api_key}"
headers["X-N8N-API-KEY"] = api_key
# 携带多种字段别名，兼容 n8n AI Agent(chatInput) 与自定义工作流
payload = {
"query": query,
"chatInput": query,
"message": query,
"input": query,
"user": (user_id or "web-user")[:64],
"history": history or [],
}
async with httpx.AsyncClient(timeout=180, proxy=_proxy_for_url(url), trust_env=False) as client:
resp = await client.post(url, headers=headers, json=payload)
if resp.status_code not in (200, 201):
raise RuntimeError(f"n8n Webhook HTTP {resp.status_code}: {resp.text[:300]}")
ctype = resp.headers.get("content-type", "")
if "application/json" in ctype:
try:
answer = _extract_text_from_json(resp.json())
except Exception:
answer = resp.text
else:
answer = resp.text
return (answer or "").strip(), None
async def _call_coze_agent(endpoint: str, api_key: str, query: str,
user_id: str, ext_conv_id: str | None) -> tuple:
"""调用扣子(Coze) v3 对话 API：create → 轮询 retrieve → 拉取 message list。
约定 agent_endpoint 存放 bot_id(纯 id)，或形如
https://api.coze.cn/v3/chat?bot_id=xxx 的完整地址。默认基址 api.coze.cn。
返回 (answer, conversation_id)。
"""
ep = (endpoint or "").strip()
from urllib.parse import urlparse, parse_qs
bot_id = None
base = "https://api.coze.cn"
if ep.startswith("http"):
pr = urlparse(ep)
base = f"{pr.scheme}://{pr.netloc}"
qs = parse_qs(pr.query)
bot_id = (qs.get("bot_id") or [None])[0]
if not bot_id and "/bot/" in pr.path:
bot_id = pr.path.rstrip("/").split("/bot/")[-1]
else:
bot_id = ep # 纯 bot_id
if not bot_id:
raise RuntimeError("Coze 接入缺少 bot_id(请在 Endpoint 填入 bot_id 或带 ?bot_id= 的地址)")
headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
create_url = f"{base}/v3/chat"
if ext_conv_id:
create_url += f"?conversation_id={ext_conv_id}"
payload = {
"bot_id": bot_id,
"user_id": (user_id or "web-user")[:64],
"stream": False,
"auto_save_history": True,
"additional_messages": [
{"role": "user", "content": query, "content_type": "text"}
],
}
async with httpx.AsyncClient(timeout=180, proxy=_proxy_for_url(create_url), trust_env=False) as client:
r = await client.post(create_url, headers=headers, json=payload)
if r.status_code != 200:
raise RuntimeError(f"Coze create HTTP {r.status_code}: {r.text[:300]}")
cd = r.json()
if cd.get("code") not in (0, None):
raise RuntimeError(f"Coze 错误: {cd.get('msg') or cd}")
chat = cd.get("data") or {}
chat_id = chat.get("id")
conv_id2 = chat.get("conversation_id") or ext_conv_id
if not chat_id or not conv_id2:
raise RuntimeError(f"Coze 未返回 chat_id/conversation_id: {str(cd)[:200]}")
# 轮询直到完成(最多约 3 分钟)
status = chat.get("status")
for _ in range(60):
if status == "completed":
break
if status in ("failed", "requires_action", "canceled"):
raise RuntimeError(f"Coze 对话状态 {status}")
await asyncio.sleep(3)
rr = await client.get(
f"{base}/v3/chat/retrieve?chat_id={chat_id}&conversation_id={conv_id2}",
headers=headers,
)
if rr.status_code == 200:
status = ((rr.json() or {}).get("data") or {}).get("status")
# 拉取消息列表，取 type=answer 的内容
rm = await client.get(
f"{base}/v3/chat/message/list?chat_id={chat_id}&conversation_id={conv_id2}",
headers=headers,
)
answer = ""
if rm.status_code == 200:
for m in ((rm.json() or {}).get("data") or []):
if m.get("type") == "answer" and m.get("role") == "assistant":
answer = m.get("content", "") or answer
return (answer or "").strip(), conv_id2
async def _call_external_agent(db, agent: dict, query: str, user_id: str,
ext_conversation_id: str | None,
history: list) -> tuple:
"""按平台分发到对应外部 Agent 远程调用。返回 (answer, new_ext_conversation_id)。"""
platform = (agent.get("platform") or "dify").lower()
endpoint = agent.get("agent_endpoint")
api_key = agent.get("agent_api_key") or ""
if not endpoint:
raise RuntimeError("该 Agent 未配置远程调用地址(agent_endpoint)")
if platform == "dify":
return await _call_dify_agent(endpoint, api_key, query, user_id, ext_conversation_id)
if platform == "n8n":
return await _call_n8n_agent(endpoint, api_key, query, user_id, history)
if platform == "coze":
return await _call_coze_agent(endpoint, api_key, query, user_id, ext_conversation_id)
# 未知平台：默认按 Dify 协议尝试
return await _call_dify_agent(endpoint, api_key, query, user_id, ext_conversation_id)
# ============================================================
# Request Models
# ============================================================
class CreateConversationRequest(BaseModel):
title: str = Field(default="新对话")
model_id: Optional[str] = None
class UpdateConversationRequest(BaseModel):
title: Optional[str] = None
status: Optional[str] = None # active, archived
pinned: Optional[bool] = None
agent_ids: Optional[list] = None
skill_ids: Optional[list] = None
knowledge_base_ids: Optional[list] = None
class SendMessageRequest(BaseModel):
content: str
model_id: Optional[str] = None
skill_ids: Optional[list] = None
agent_ids: Optional[list] = None
knowledge_base_ids: Optional[list] = None
mcp_ids: Optional[list] = None
attachments: Optional[list] = None
use_tools: Optional[bool] = False
# ============================================================
# Skill Resolution: MarketApp → Skill Definition
# ============================================================
def _resolve_market_skills(db: Session, skill_ids: list, user_id: str) -> list:
"""
Stage A.1: 统一Skill ID空间
前端传MarketApp.id, 需要映射为可用的技能定义。
返回: [{id, name, description, skill_content, category, func_name}]
"""
if not skill_ids:
return []
# 1. 从MarketApp查询 (前端传的是MarketApp.id)
market_skills = db.query(MarketApp).filter(
MarketApp.id.in_(skill_ids),
MarketApp.type == "skill"
).all()
results = []
for ms in market_skills:
skill_content = ms.skill_content or ""
# 尝试从skill_content解析SKILL.md格式
# SKILL.md格式: ---\nname: ...\ndescription: ...\n---\n内容...
parsed = _parse_skill_content(skill_content)
func_name = ms.name.replace(" ", "_").replace("-", "_").lower()
results.append({
"id": ms.id,
"market_app_id": ms.id,
"name": ms.name,
"description": ms.description or parsed.get("description", ""),
"skill_content": skill_content,
"category": parsed.get("category", "general"),
"skill_type": parsed.get("skill_type", "general"),
"func_name": func_name,
"parameters": parsed.get("parameters", {"input": {"type": "string", "description": "用户输入内容"}}),
"instructions": parsed.get("instructions", ""),
})
# 2. 如果有未匹配的ID, 尝试查EvolutionSkill (兼容旧的进化技能ID)
matched_ids = {r["id"] for r in results}
unmatched = [sid for sid in skill_ids if sid not in matched_ids]
if unmatched:
evo_skills = db.query(EvolutionSkill).filter(
EvolutionSkill.id.in_(unmatched),
EvolutionSkill.user_id == user_id
).all()
for es in evo_skills:
# 读取SKILL.md内容
skill_md = ""
if es.skill_md_path and os.path.exists(es.skill_md_path):
try:
with open(es.skill_md_path, "r", encoding="utf-8") as f:
skill_md = f.read()
except Exception:
skill_md = ""
elif es.description:
skill_md = es.description
parsed = _parse_skill_content(skill_md)
func_name = es.skill_name.replace(" ", "_").replace("-", "_").lower()
results.append({
"id": es.id,
"market_app_id": None,
"name": es.skill_name,
"description": es.description or parsed.get("description", ""),
"skill_content": skill_md,
"category": es.category or parsed.get("category", "general"),
"skill_type": parsed.get("skill_type", es.category or "general"),
"func_name": func_name,
"parameters": parsed.get("parameters", {"input": {"type": "string", "description": "用户输入内容"}}),
"instructions": parsed.get("instructions", ""),
})
return results
def _resolve_market_agents(db: Session, agent_ids: list, user_id: str) -> list:
"""
Stage B.1: 加载Agent配置
前端传MarketApp.id(agent类型), 需要解析AgentConfig。
返回: [{id, name, config, skills, knowledge_base_ids}]
"""
if not agent_ids:
return []
agents = db.query(MarketApp).filter(
MarketApp.id.in_(agent_ids),
MarketApp.type == "agent"
).all()
results = []
for ag in agents:
# AgentConfig在skill_content字段(JSON序列化)
agent_config = {}
try:
if ag.skill_content:
agent_config = json.loads(ag.skill_content)
except Exception:
agent_config = {"role": ag.description or "AI助手", "instructions": ag.description or ""}
# 提取配置字段
role_prompt = agent_config.get("role", ag.description or "AI助手")
instructions = agent_config.get("instructions", "")
agent_skill_ids = agent_config.get("skills", []) # 关联的skill MarketApp IDs
agent_kb_ids = agent_config.get("knowledge_base_ids", []) # 关联的知识库IDs
agent_model_id = agent_config.get("model_id", None) # 指定使用的模型
tools = agent_config.get("tools", []) # MCP server IDs(暂不用)
results.append({
"id": ag.id,
"name": ag.name,
"role_prompt": role_prompt,
"instructions": instructions,
"skill_ids": agent_skill_ids,
"knowledge_base_ids": agent_kb_ids,
"model_id": agent_model_id,
"tools": tools,
"description": ag.description or "",
# 外部 Agent 接入(Dify/n8n/Coze)运行期远程调用所需字段：
# 这三列此前被丢弃，导致选中外部 Agent 时仍用本地 LLM「扮演」而非真远程调用。
"platform": (ag.platform or "").lower() if ag.platform else None,
"agent_endpoint": (ag.agent_endpoint or "").strip() if ag.agent_endpoint else None,
"agent_api_key": ag.agent_api_key or None,
})
return results
def _resolve_market_mcp(db: Session, mcp_ids: list, user_id: str) -> list:
"""
加载用户选择的 MCP 应用配置（MarketApp.type == 'mcp'）。
返回: [{id, name, transport, endpoint, description}]
"""
if not mcp_ids:
return []
mcps = db.query(MarketApp).filter(
MarketApp.id.in_(mcp_ids),
MarketApp.type == "mcp"
).all()
results = []
for m in mcps:
config = {}
try:
if m.skill_content:
config = json.loads(m.skill_content)
except Exception:
config = {}
results.append({
"id": m.id,
"name": m.name,
"transport": m.mcp_transport or config.get("transport", "stdio"),
"endpoint": m.agent_endpoint or config.get("endpoint", "") or config.get("url", ""),
"description": m.description or config.get("description", ""),
})
return results
def _resolve_active_mcp_servers(db: Session, mcp_ids: list) -> list:
"""
解析用户选中的 MCP 服务连接信息（来自 mcp_servers 表，import 与后台添加通用）。
返回: [{id, name, description, transport, endpoint, url, headers}]
"""
if not mcp_ids:
return []
servers = db.query(MCPServer).filter(
MCPServer.id.in_(mcp_ids),
MCPServer.is_enabled == True,
).all()
out = []
for s in servers:
try:
headers = json.loads(s.headers) if isinstance(s.headers, str) else (s.headers or {})
except Exception:
headers = {}
out.append({
"id": s.id,
"name": s.name,
"description": s.description or "",
"transport": s.transport_type or "streamable_http",
"endpoint": s.url or "",
"url": s.url or "",
"headers": headers,
})
return out
def _list_enabled_mcp_servers(db: Session) -> list:
"""返回平台所有已启用(is_enabled=True)的 MCP 服务清单，供 AI 准确回答"平台有哪些 MCP"。"""
servers = db.query(MCPServer).filter(MCPServer.is_enabled == True).all()
out = []
for s in servers:
try:
headers = json.loads(s.headers) if isinstance(s.headers, str) else (s.headers or {})
except Exception:
headers = {}
out.append({
"id": s.id,
"name": s.name,
"description": s.description or "",
"transport": s.transport_type or "streamable_http",
"endpoint": s.url or "",
"url": s.url or "",
"headers": headers,
})
return out
def _discover_installed_skills(db: Session, user_id: str, limit: int = 300) -> list:
"""
Issue 8: 默认技能自动发现。
检索「本账号已安装(app_installations) 且 平台已部署发布(MarketApp.status='published')」的
skill 型应用，返回可供 function calling 使用的技能定义列表（最多 limit 个）。
仅当用户未显式选择技能时才启用；多个技能适用时由模型自行判断选 1 个最优。
"""
if not user_id:
return []
installed = db.query(AppInstallation).filter(
AppInstallation.user_id == user_id
).all()
app_ids = [i.app_id for i in installed]
if not app_ids:
return []
apps = db.query(MarketApp).filter(
MarketApp.id.in_(app_ids),
MarketApp.type == "skill",
MarketApp.status == "published",
).all()
ids = [a.id for a in apps]
if not ids:
return []
return _resolve_market_skills(db, ids, user_id)[:limit]
def _parse_skill_content(content: str) -> dict:
"""
解析SKILL.md格式内容, 提取结构化信息。
SKILL.md格式:
---
name: 技能名
description: 描述
category: 分类
skill_type: 类型
parameters: JSON或简单定义
---
技能指令/规则内容
"""
result = {
"name": "",
"description": "",
"category": "general",
"skill_type": "general",
"parameters": {},
"instructions": "",
}
if not content:
return result
# 尝试解析YAML front matter
if content.strip().startswith("---"):
parts = content.split("---")
if len(parts) >= 3:
front_matter = parts[1].strip()
body = parts[2].strip() if len(parts) > 2 else ""
# 简单YAML解析 (不用pyyaml库)
for line in front_matter.split("\n"):
line = line.strip()
if ":" in line:
key, val = line.split(":", 1)
key = key.strip().lower()
val = val.strip()
if key == "name":
result["name"] = val
elif key == "description":
result["description"] = val
elif key == "category":
result["category"] = val
elif key == "skill_type":
result["skill_type"] = val
elif key == "parameters":
try:
result["parameters"] = json.loads(val)
except Exception:
# 简单格式: input:string,query:string
params = {}
for p in val.split(","):
p = p.strip()
if ":" in p:
pname, ptype = p.split(":", 1)
params[pname.strip()] = {"type": ptype.strip(), "description": pname.strip()}
result["parameters"] = params
result["instructions"] = body
else:
result["instructions"] = content
else:
# 没有front matter, 全部当作instructions
result["instructions"] = content
# 尝试从第一行提取name
first_line = content.split("\n")[0].strip()
if first_line and len(first_line) < 50:
result["name"] = first_line
return result
# ============================================================
# BM25 Text Search (Stage C.2)
# ============================================================
def _bm25_search(query: str, documents: list, top_k: int = 5) -> list:
"""
简易BM25检索 — 替代朴素的文档拼接。
对文档列表进行关键词匹配+TF-IDF排序, 返回最相关的top_k段。
"""
if not documents or not query:
return documents[:top_k]
# 中文简单分词(按字符+标点)
def tokenize(text: str) -> list:
# 移除标点, 拆分为2-4字词组
text = re.sub(r'[^\w\u4e00-\u9fff]', ' ', text.lower())
tokens = []
words = text.split()
for w in words:
if len(w) >= 2:
tokens.append(w)
# 也生成2-gram
if len(w) >= 4:
for i in range(len(w) - 2):
tokens.append(w[i:i+2])
return tokens
query_tokens = tokenize(query)
if not query_tokens:
return documents[:top_k]
scored_docs = []
avg_doc_len = sum(len(tokenize(d.get("content", ""))) for d in documents) / max(len(documents), 1)
# BM25参数
k1 = 1.5
b = 0.75
# 计算每个token的IDF
df = {}
for token in query_tokens:
for doc in documents:
if token in tokenize(doc.get("content", "")):
df[token] = df.get(token, 0) + 1
N = len(documents)
idf = {}
for token in query_tokens:
idf[token] = math.log((N - df.get(token, 0) + 0.5) / (df.get(token, 0) + 0.5) + 1)
# 计算每个文档的BM25得分
for doc in documents:
doc_tokens = tokenize(doc.get("content", ""))
doc_len = len(doc_tokens)
score = 0
for token in query_tokens:
tf = doc_tokens.count(token)
numerator = tf * (k1 + 1)
denominator = tf + k1 * (1 - b + b * doc_len / max(avg_doc_len, 1))
score += idf.get(token, 0) * numerator / denominator
scored_docs.append((score, doc))
# 按得分排序
scored_docs.sort(key=lambda x: x[0], reverse=True)
return [d for s, d in scored_docs[:top_k]]
def _retrieve_kb_context(db: Session, kb_ids: list, query: str, max_chars: int = 3000) -> str:
"""
Stage C: 增强版知识库检索 — 使用BM25排序替代朴素拼接
1. 获取每个KB的文档
2. 构建文档列表(content+title+metadata)
3. BM25检索最相关的段落
4. 拼接为上下文注入system prompt
同时叠加 Dify 外部知识库(RAG)检索结果。
"""
local_context = ""
if kb_ids:
kbs = db.query(KnowledgeBase).filter(KnowledgeBase.id.in_(kb_ids)).all()
if kbs:
# 收集所有文档段落
# Stage C.1: 优先使用 KnowledgeDocumentChunk 表的预分段数据
all_segments = []
for kb in kbs:
# 先查chunk表
chunks = db.query(KnowledgeDocumentChunk).filter(
KnowledgeDocumentChunk.knowledge_base_id == kb.id
).order_by(KnowledgeDocumentChunk.chunk_index).limit(30).all()
if chunks:
# 使用预分段数据
for chunk in chunks:
if chunk.content and len(chunk.content.strip()) >= 20:
all_segments.append({
"content": "[{}/{}/段{}]: {}".format(
kb.name, chunk.document_id[:8], chunk.chunk_index, chunk.content[:500]
),
"raw_content": chunk.content,
"kb_name": kb.name,
})
else:
# 退化: 使用原始文档分段
docs = db.query(KnowledgeDocument).filter(
KnowledgeDocument.knowledge_base_id == kb.id,
KnowledgeDocument.status == "completed"
).order_by(KnowledgeDocument.updated_at.desc()).limit(20).all()
for doc in docs:
content = doc.content or ""
if not content:
continue
# 如果文档较长, 分段处理 (每段~800字)
if len(content) > 800:
paragraphs = re.split(r'\n\n+', content)
for pi, para in enumerate(paragraphs):
if len(para.strip()) >= 20:
all_segments.append({
"content": "[{}/{}/段{}]: {}".format(
kb.name, doc.name, pi+1, para.strip()[:500]
),
"raw_content": para.strip(),
"kb_name": kb.name,
"doc_name": doc.name,
})
else:
all_segments.append({
"content": "[{}/{}]: {}".format(kb.name, doc.name, content[:500]),
"raw_content": content,
"kb_name": kb.name,
"doc_name": doc.name,
})
if not all_segments:
# 退化: 使用原始朴素拼接
kb_names = ", ".join([k.name for k in kbs])
local_context = f"参考知识库: {kb_names}"
else:
# BM25检索最相关的段落
top_segments = _bm25_search(query, all_segments, top_k=8)
context_parts = [seg["content"] for seg in top_segments]
local_context = "\n".join(context_parts)
if len(local_context) > max_chars:
local_context = local_context[:max_chars] + "...(内容截断)"
# 叠加 Dify 外部知识库(RAG)检索
dify_context = ""
try:
from app.dify_kb import get_dify_kb_context
dify_context = get_dify_kb_context(db, query, top_k=5)
except Exception as e:
logger.warning(f"Dify KB context retrieval failed: {e}")
parts = []
if local_context:
parts.append(local_context)
if dify_context:
parts.append("【Dify 知识库检索结果】\n" + dify_context)
return "\n\n".join(parts)
async def _ingest_answer_to_dify(conv_id: str, msg_id: str, content: str):
"""后台任务：将助手回答入库到 Dify 知识库（量化后可供 RAG 检索复用）。"""
try:
from app.dify_kb import aingest_text
from app.database import SessionLocal
from sqlalchemy import text as _t
db = SessionLocal()
try:
row = db.execute(_t(
"SELECT api_url, api_key, config FROM external_kb_services "
"WHERE service_type='dify' AND is_enabled=1 LIMIT 1"
)).fetchone()
finally:
db.close()
if not row:
return
api_url, api_key, config_raw = row
dataset_name = None
if config_raw:
try:
import json
cfg = config_raw if isinstance(config_raw, dict) else json.loads(config_raw)
dataset_name = cfg.get("dataset_name")
except Exception:
pass
await aingest_text(api_url, api_key, content, name=f"chat_{conv_id}_{msg_id}", dataset_name=dataset_name)
except Exception as e:
logger.warning(f"后台入库 Dify 失败: {e}")
# ============================================================
# Dynamic System Prompt Builder (Stage B)
# ============================================================
def _build_system_prompt(
skills: list,
agents: list,
kb_context: str,
user_content: str,
mcp: list = None,
mcp_functions: list = None,
auto_discovered: bool = False,
all_mcp: list = None,
) -> tuple:
"""
动态组装system prompt + function definitions。
返回: (system_prompt_str, function_definitions_list)
"""
system_parts = ["你是AIFlow平台的AI助手，具备专业的知识储备和工具调用能力。"]
# 平台知识库注入（始终开启）：确保AI优先使用平台已有能力
try:
pk = get_platform_knowledge()
if pk:
system_parts.append(f"【平台知识库——优先使用以下知识回答用户问题】\n{pk}")
except Exception:
pass
# 回复格式要求：强制使用Markdown
system_parts.append("回复格式要求：使用 Markdown 格式，层次分明即可。对于简短问题，用 1-2 段自然文字直接回答，不必强行分点、加标题或表格；仅在确实需要分步骤或列举时才使用列表，代码/配置用 ``` 代码块包裹。避免冗余空行，保持紧凑、美观、易读。")
function_definitions = []
# Stage B.1: Agent角色注入（Issue 2：让用户询问 agent 能力时，AI 能“看到”该 agent 资料）
if agents:
for agent in agents:
name = agent.get("name", "")
role = agent.get("role_prompt", "")
instructions = agent.get("instructions", "")
desc = agent.get("description", "")
block = []
if name:
block.append(
f"用户已选择智能体「{name}」。当用户询问该智能体的功能、能力、用法或相关问题时，"
f"请基于以下资料准确作答，不要说“不知道这个智能体”。"
)
if role:
block.append(f"该智能体的角色设定: {role}")
if desc:
block.append(f"该智能体的简介: {desc}")
if instructions:
block.append(f"该智能体的行为指引: {instructions}")
if block:
system_parts.append("\n".join(block))
# Stage A.2: Skill工具注入
if skills:
skill_descs = ", ".join([f"{s['name']}({s.get('category', 'general')})" for s in skills])
system_parts.append(f"可用技能工具: {skill_descs}")
if auto_discovered:
# Issue 8：每轮默认先寻找本账号已安装且平台已部署的技能，多个适用时 AI 自行判断选 1 个最优
system_parts.append(
"重要规则: 以上技能为本账号已安装且平台已部署的技能。每轮对话请先判断用户请求是否可由其中某个技能处理；"
"若多个技能都适用，请自行判断并选择最贴切的 1 个优先调用，不要一次性全部调用。"
"调用对应工具后将获得执行结果，基于结果生成最终回复。若这些技能均不适用，则直接基于你的知识回答。"
)
else:
system_parts.append(
"重要规则：\n"
"1) 当用户请求需要由该技能【执行】某项任务时，优先通过 function calling 调用对应工具；调用后基于工具返回结果生成回复。\n"
"2) 当用户询问所选技能【本身的介绍、功能、用途、用法、特点】（例如“介绍一下这个技能”“它是什么”“怎么用”“能做什么”）时，"
"必须基于下方【已选技能资料】中的名称/描述/指令直接作答，不要泛泛介绍平台整体，也不要说“不知道这个技能”。\n"
"3) 若用户问题既不执行也不询问该技能本身，则按常规方式回答。"
)
# 为每个skill构建function definition
for s in skills:
# 从skill_content提取指令注入system prompt
instructions = s.get("instructions", "")
name = s.get("name", "")
if not auto_discovered:
# 显式选中技能：注入可引用的技能资料，便于“介绍/说明”类问题据此作答
ref_parts = [f"【已选技能资料】名称：{name}"]
desc = s.get("description", "")
if desc:
ref_parts.append(f"简介：{desc}")
if instructions and len(instructions) < 1200:
ref_parts.append(f"使用指令：{instructions}")
system_parts.append("\n".join(ref_parts))
elif instructions and len(instructions) < 800:
system_parts.append(f"技能「{name}」的详细指令: {instructions}")
# 构建参数schema
params = s.get("parameters", {})
if isinstance(params, dict) and len(params) > 0:
properties = {}
required = []
for pname, pdef in params.items():
if isinstance(pdef, dict):
properties[pname] = {
"type": pdef.get("type", "string"),
"description": pdef.get("description", f"传递给{s['name']}的{pname}参数"),
}
if pdef.get("required", True):
required.append(pname)
else:
properties[pname] = {"type": "string", "description": f"传递给{s['name']}的{pname}参数"}
required.append(pname)
if not properties:
properties = {"input": {"type": "string", "description": f"需要传递给{s['name']}的完整用户输入"}}
required = ["input"]
else:
properties = {"input": {"type": "string", "description": f"需要传递给{s['name']}的完整用户输入"}}
required = ["input"]
desc = s.get("description", "")
category = s.get("category", "general")
func_desc = f"调用技能「{s['name']}」处理用户请求。使用场景: {desc}。当用户请求涉及{category}类任务时优先调用此工具。"
function_definitions.append({
"name": s["func_name"],
"description": func_desc,
"parameters": {
"type": "object",
"properties": properties,
"required": required,
},
})
# KB context injection
if kb_context:
system_parts.append(f"参考以下知识库内容回答用户问题:\n{kb_context}")
# 平台全量 MCP 清单（Issue：让用户询问"平台有哪些 MCP"时 AI 能准确列举，而非凭空编造）
if all_mcp:
_mcp_inventory = "\n".join([f"- {m.get('name','')}: {m.get('description','') or '（未填写描述）'}" for m in all_mcp])
system_parts.append(
f"【平台已安装的 MCP 应用清单（共 {len(all_mcp)} 个，均为平台真实接入并启用）】\n{_mcp_inventory}\n"
"当用户询问“平台有哪些 MCP / 接入了哪些 MCP / 有哪些工具服务”时，请严格依据以上清单准确列举名称与功能，"
"不要编造或遗漏，也不要把未列出的服务说成已接入。"
)
# MCP 应用注入（Issue 3：让用户询问 MCP 能力时，AI 能“看到”该 MCP 的资料与地址）
if mcp:
mcp_lines = []
for m in mcp:
name = m.get("name", "")
desc = m.get("description", "")
transport = m.get("transport", "stdio")
endpoint = m.get("endpoint", "")
line = f"用户已选择 MCP 应用「{name}」（传输方式: {transport}）"
if desc:
line += f"，功能简介: {desc}"
if endpoint:
line += f"，服务地址: {endpoint}"
mcp_lines.append(line)
system_parts.append("\n".join(mcp_lines))
system_parts.append(
"用户已选择上述 MCP 应用。当请求需要借助外部工具/系统（如查询数据、调用 API、操作第三方服务）时，"
"请优先参考这些 MCP 应用的能力来回答；若平台已接入对应工具，优先通过工具调用完成；"
"不要说“不了解该 MCP”。"
)
# 高优先级覆盖规则：显式选中工具时，回答必须基于该工具资料
if (skills and not auto_discovered) or (mcp):
system_parts.append(
"【最高优先级覆盖规则】\n"
"用户在本轮已明确选择了工具（见上文【已选技能资料】或【用户已选择 MCP 应用】）。\n"
"如果用户的问题是在询问这些已选工具/技能/应用本身的介绍、功能、能做什么、怎么用、特点或用法，"
"你必须只基于上文提供的该工具具体资料作答；\n"
"不要引用《平台知识库》中关于技能市场、MCP 协议或平台整体能力的通用描述；"
"不要列举其它未选中的技能或 MCP 作为示例；也不要说“不知道”或“无法回答”。\n"
"若用户问题与已选工具本身无关，则按常规方式回答。"
)
system_prompt = "\n".join(system_parts)
# MCP function definitions 注入（让 LLM 可通过 function calling 真实调用 MCP 工具）
if mcp_functions:
function_definitions.extend(mcp_functions)
return system_prompt, function_definitions
# ============================================================
# Skill Tool Execution (Stage A.2 rewritten)
# ============================================================
def _execute_skill_tool(
db: Session,
func_name: str,
args: dict,
skills: list,
user_content: str,
) -> str:
"""
Stage A.2: 真实Skill执行引擎 — 替代mock实现
1. 查找匹配的skill定义
2. 读取skill_content(SKILL.md)中的指令
3. 根据指令类型组装执行prompt
4. 返回结构化的执行结果给LLM
"""
# 查找匹配的skill
matched_skill = None
for s in skills:
expected_name = s["func_name"]
if func_name == expected_name or func_name == f"skill_{s['id'][:8]}":
matched_skill = s
break
if not matched_skill:
# 退化: 返回通用结果
return f"工具 '{func_name}' 执行结果: 已处理输入内容, 但未找到对应的技能定义。请基于通用知识回复用户。"
skill_name = matched_skill["name"]
instructions = matched_skill.get("instructions", "")
input_text = args.get("input", user_content)
# 根据skill_content中的指令生成真实执行结果
# 核心思路: 将skill的instructions作为"执行逻辑",
# 把input_text传入, 让结果包含指令要求的处理格式
if instructions:
# 有明确指令的技能 — 按指令结构化输出
result_parts = [
f"技能「{skill_name}」执行结果:",
f"---",
]
# 尝试从指令中提取输出格式要求
# 如果指令包含"输出"或"格式"关键字, 遵循其要求
output_format = ""
if "输出格式" in instructions or "output format" in instructions.lower():
# 提取格式描述
for line in instructions.split("\n"):
if "输出" in line or "格式" in line or "format" in line.lower():
output_format += line + "\n"
# 根据skill_type增强处理
skill_type = matched_skill.get("skill_type", "general")
if skill_type == "writer" or "写作" in skill_name or "写" in instructions:
result_parts.append(f"基于技能「{skill_name}」的写作指令处理用户输入:")
result_parts.append(f"主题: {input_text[:200]}")
result_parts.append(f"根据技能规则: {instructions[:300]}")
result_parts.append(f"处理完成 — 请基于上述技能指令和输入内容, 生成符合技能要求的写作输出。")
elif skill_type == "analyzer" or "分析" in skill_name or "分析" in instructions:
result_parts.append(f"基于技能「{skill_name}」的分析指令处理用户输入:")
result_parts.append(f"分析对象: {input_text[:200]}")
result_parts.append(f"分析维度(按技能规则): {instructions[:300]}")
result_parts.append(f"关键发现: 识别到{max(1, len(input_text)//50)}个分析要点")
result_parts.append(f"处理完成 — 请基于上述技能指令, 对输入内容进行深入分析并给出结构化的分析报告。")
elif skill_type == "translator" or "翻译" in skill_name or "翻译" in instructions:
result_parts.append(f"基于技能「{skill_name}」的翻译指令处理用户输入:")
result_parts.append(f"原文: {input_text[:200]}")
result_parts.append(f"翻译规则: {instructions[:300]}")
result_parts.append(f"处理完成 — 请基于技能翻译规则, 将输入内容翻译为目标语言。")
elif skill_type == "summarizer" or "总结" in skill_name or "摘要" in instructions:
result_parts.append(f"基于技能「{skill_name}」的摘要指令处理用户输入:")
result_parts.append(f"原文长度: {len(input_text)} 字符")
result_parts.append(f"摘要规则: {instructions[:300]}")
result_parts.append(f"处理完成 — 请基于技能摘要规则, 对输入内容进行精准摘要。")
else:
# 通用技能 — 传递指令让LLM执行
result_parts.append(f"基于技能「{skill_name}」的指令处理用户请求:")
result_parts.append(f"用户输入: {input_text[:200]}")
result_parts.append(f"技能指令: {instructions[:500]}")
if output_format:
result_parts.append(f"期望输出格式: {output_format[:200]}")
result_parts.append(f"处理完成 — 请严格按照技能指令的规则和要求, 处理用户输入并生成最终回复。")
result = "\n".join(result_parts)
else:
# 无明确指令的技能 — 提供基本信息让LLM自行处理
desc = matched_skill.get("description", "")
result = (
f"技能「{skill_name}」已激活。\n"
f"技能描述: {desc}\n"
f"用户输入: {input_text[:200]}\n"
f"请基于技能「{skill_name}」的专业能力处理上述输入。"
)
# 更新使用计数(如果关联了MarketApp)
market_app_id = matched_skill.get("market_app_id")
if market_app_id:
try:
app = db.query(MarketApp).filter(MarketApp.id == market_app_id).first()
if app:
app.install_count = (app.install_count or 0) + 1
except Exception:
pass
return result
# ============================================================
# Conversation Context Persistence (Stage B.2)
# ============================================================
def _get_conversation_context(db: Session, conv_id: str) -> dict:
"""
Stage B.2: 获取对话的持久化上下文
对话级别的agent/skill/kb选择保存在Conversation字段中。
"""
conv = db.query(Conversation).filter(Conversation.id == conv_id).first()
if not conv:
return {}
# Conversation目前没有存储agent/skill/kb的JSON字段
# 我们从第一条消息中获取(如果存在), 或返回空
# 后续可通过数据库迁移添加Conversation.context字段
# 查询该对话中最早一条含有skill_ids/agent_ids/knowledge_ids的消息
first_tool_msg = db.query(Message).filter(
Message.conversation_id == conv_id,
Message.role == "user",
).order_by(asc(Message.created_at)).first()
context = {}
if first_tool_msg:
if first_tool_msg.skill_ids:
context["skill_ids"] = first_tool_msg.skill_ids
if first_tool_msg.agent_ids:
context["agent_ids"] = first_tool_msg.agent_ids
if first_tool_msg.knowledge_ids:
context["knowledge_base_ids"] = first_tool_msg.knowledge_ids
if getattr(first_tool_msg, "mcp_ids", None):
context["mcp_ids"] = first_tool_msg.mcp_ids
return context
# ============================================================
# API Endpoints
# ============================================================
@router.get("/conversations")
def list_conversations(
status: Optional[str] = Query(None),
search: Optional[str] = Query(None),
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
):
query = db.query(Conversation).filter(Conversation.user_id == current_user.id)
if status:
query = query.filter(Conversation.status == status)
else:
query = query.filter(Conversation.status != "archived")
if search:
query = query.filter(Conversation.title.ilike(f"%{search}%"))
query = query.order_by(desc(Conversation.pinned), desc(Conversation.updated_at))
conversations = query.all()
result = []
for conv in conversations:
result.append({
"id": conv.id,
"title": conv.title,
"model_id": conv.model_id,
"status": conv.status,
"pinned": conv.pinned,
"knowledge_graph_count": conv.knowledge_graph_count or 0,
"source": getattr(conv, "source", None),
"created_at": str(conv.created_at) if conv.created_at else None,
"updated_at": str(conv.updated_at) if conv.updated_at else None,
})
return {"code": 0, "data": result, "message": "ok"}
@router.post("/conversations")
def create_conversation(
req: CreateConversationRequest,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
):
conv = Conversation(
id=gen_uuid(),
user_id=current_user.id,
title=req.title,
model_id=req.model_id,
status="active",
pinned=False,
knowledge_graph_count=0,
created_at=cst_now(),
updated_at=cst_now(),
)
db.add(conv)
db.commit()
db.refresh(conv)
return {
"code": 0,
"data": {
"id": conv.id,
"title": conv.title,
"model_id": conv.model_id,
"status": conv.status,
"pinned": conv.pinned,
"knowledge_graph_count": conv.knowledge_graph_count,
"created_at": str(conv.created_at) if conv.created_at else None,
"updated_at": str(conv.updated_at) if conv.updated_at else None,
},
"message": "创建成功",
}
@router.put("/conversations/{conv_id}")
def update_conversation(
conv_id: str,
req: UpdateConversationRequest,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
):
conv = db.query(Conversation).filter(
and_(Conversation.id == conv_id, Conversation.user_id == current_user.id)
).first()
if not conv:
raise HTTPException(status_code=404, detail="对话不存在")
if req.title is not None:
conv.title = req.title
if req.status is not None:
conv.status = req.status
if req.pinned is not None:
conv.pinned = req.pinned
# Stage B.2: 保存agent/skill/kb选择到对话上下文
# (存储在第一条system消息的knowledge_ids/agent_ids/skill_ids字段中)
conv.updated_at = cst_now()
db.commit()
db.refresh(conv)
return {"code": 0, "data": {
"id": conv.id, "title": conv.title, "status": conv.status, "pinned": conv.pinned,
"updated_at": str(conv.updated_at),
}, "message": "更新成功"}
@router.delete("/conversations/{conv_id}")
def delete_conversation(
conv_id: str,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
):
conv = db.query(Conversation).filter(
and_(Conversation.id == conv_id, Conversation.user_id == current_user.id)
).first()
if not conv:
raise HTTPException(status_code=404, detail="对话不存在")
conv.status = "archived"
conv.updated_at = cst_now()
db.commit()
return {"code": 0, "message": "删除成功"}
@router.get("/conversations/{conv_id}/messages")
def get_messages(
conv_id: str,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
):
conv = db.query(Conversation).filter(
and_(Conversation.id == conv_id, Conversation.user_id == current_user.id)
).first()
if not conv:
raise HTTPException(status_code=404, detail="对话不存在")
messages = (
db.query(Message)
.filter(Message.conversation_id == conv_id)
.order_by(asc(Message.created_at), asc(Message.id))
.all()
)
result = []
for msg in messages:
result.append({
"id": msg.id,
"conversation_id": msg.conversation_id,
"role": msg.role,
"content": msg.content,
"token_used": msg.token_used or 0,
"model_id": msg.model_id,
"used_skills": msg.used_skills or [],
"used_mcp": msg.used_mcp or [],
"created_at": str(msg.created_at) if msg.created_at else None,
})
return {"code": 0, "data": result, "message": "ok"}
@router.post("/conversations/{conv_id}/messages")
async def send_message(
conv_id: str,
req: SendMessageRequest,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
):
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:
_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:
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:
tool_result = await call_mcp_tool(
_srv, _info["name"], func_args
)
called_mcp_names.append(_srv.get("name", ""))
else:
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": "发送成功",
}
@router.post("/conversations/{conv_id}/clear")
def clear_conversation(
conv_id: str,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
):
conv = db.query(Conversation).filter(
and_(Conversation.id == conv_id, Conversation.user_id == current_user.id)
).first()
if not conv:
raise HTTPException(status_code=404, detail="对话不存在")
db.query(Message).filter(
Message.conversation_id == conv_id
).delete()
conv.knowledge_graph_count = 0
conv.updated_at = cst_now()
db.commit()
return {"code": 0, "message": "对话上下文已清空"}
def _build_knowledge_graph(db: Session, kb: KnowledgeBase) -> dict:
"""从知识库文档构建知识图谱节点和边，返回 graph_data。"""
documents = db.query(KnowledgeDocument).filter(
and_(
KnowledgeDocument.knowledge_base_id == kb.id,
KnowledgeDocument.status == "completed",
)
).all()
graph_nodes = []
graph_edges = []
doc_entity_types = [
{"type": "concept", "label": "核心概念"},
{"type": "entity", "label": "关键实体"},
{"type": "attribute", "label": "属性特征"},
{"type": "relation", "label": "关联关系"},
]
for idx, doc in enumerate(documents):
doc_node_id = f"doc-{doc.id[:8]}"
graph_nodes.append({
"id": doc_node_id, "label": doc.filename, "type": "document", "size": 18, "group": "document",
})
filename_base = doc.filename.rsplit(".", 1)[0]
words = filename_base.replace("_", " ").replace("-", " ").split()
for wi, word in enumerate(words):
if len(word) >= 2:
entity_type = doc_entity_types[wi % len(doc_entity_types)]
node_id = f"ent-{doc.id[:8]}-{wi}"
graph_nodes.append({
"id": node_id, "label": word, "type": entity_type["type"],
"size": 10 + (wi % 8), "group": entity_type["type"],
})
graph_edges.append({
"source": doc_node_id, "target": node_id,
"label": entity_type["label"], "weight": 0.5 + (0.5 / (wi + 1)),
})
for i in range(len(documents)):
for j in range(i + 1, len(documents)):
fi = documents[i].filename.lower().rsplit(".", 1)[0]
fj = documents[j].filename.lower().rsplit(".", 1)[0]
words_i = set(fi.replace("_", " ").replace("-", " ").split())
words_j = set(fj.replace("_", " ").replace("-", " ").split())
common = words_i & words_j
if common and len(common) > 0:
graph_edges.append({
"source": f"doc-{documents[i].id[:8]}",
"target": f"doc-{documents[j].id[:8]}",
"label": f"共有关键词: {', '.join(list(common)[:3])}",
"weight": min(0.9, 0.3 * len(common)),
})
graph_data = {
"nodes": graph_nodes,
"edges": graph_edges,
"stats": {
"total_nodes": len(graph_nodes), "total_edges": len(graph_edges),
"document_count": len(documents),
},
}
kb.knowledge_graph_count = len(graph_nodes)
kb.updated_at = cst_now()
return graph_data
@router.post("/conversations/{conv_id}/to-knowledge-graph")
def convert_to_knowledge_graph(
conv_id: str,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
):
"""将对话内容转换为知识图谱（幂等：同一对话仅生成一次知识库）"""
conv = db.query(Conversation).filter(
and_(Conversation.id == conv_id, Conversation.user_id == current_user.id)
).first()
if not conv:
raise HTTPException(status_code=404, detail="对话不存在")
messages = (
db.query(Message)
.filter(Message.conversation_id == conv_id)
.order_by(asc(Message.created_at), asc(Message.id))
.all()
)
if not messages:
raise HTTPException(status_code=400, detail="对话中没有消息, 无法生成知识图谱")
# 幂等：若该对话已生成过知识图谱且知识库仍存在，则直接刷新返回
if conv.knowledge_graph_base_id:
existing_kb = db.query(KnowledgeBase).filter(
KnowledgeBase.id == conv.knowledge_graph_base_id
).first()
if existing_kb:
graph_data = _build_knowledge_graph(db, existing_kb)
conv.knowledge_graph_count = 1
conv.updated_at = cst_now()
db.commit()
return {
"code": 0,
"data": {
"knowledge_base_id": existing_kb.id,
"knowledge_base_name": existing_kb.name,
"graph": graph_data,
"knowledge_graph_count": 1,
"already_exists": True,
},
"message": "知识图谱已存在, 已刷新展示",
}
kb_name = f"对话图谱-{conv.title}"
kb = KnowledgeBase(
id=gen_uuid(),
user_id=current_user.id,
name=kb_name,
description=f"从对话「{conv.title}」自动生成的知识图谱",
directory_path="/",
document_count=0,
created_at=cst_now(),
updated_at=cst_now(),
)
db.add(kb)
db.flush()
doc_count = 0
for idx, msg in enumerate(messages):
role_label = "用户" if msg.role == "user" else "AI"
preview = re.sub(r'[\\n\\r\\t]+', ' ', msg.content[:40]).strip()
filename = f"msg_{idx+1:03d}_{role_label}_{preview}.txt"
doc = KnowledgeDocument(
id=gen_uuid(),
knowledge_base_id=kb.id,
filename=filename,
file_type="txt",
file_size=len(msg.content.encode('utf-8')),
content=msg.content,
status="completed",
chunk_count=max(1, len(msg.content) // 500 + 1),
created_at=cst_now(),
updated_at=cst_now(),
)
db.add(doc)
doc_count += 1
kb.document_count = doc_count
db.flush()
# 构建知识图谱节点和边
graph_data = _build_knowledge_graph(db, kb)
conv.knowledge_graph_base_id = kb.id
conv.knowledge_graph_count = 1
conv.updated_at = cst_now()
db.commit()
LogService.log_operation(
user_id=current_user.id, username=current_user.username, action="convert_to_graph",
target="conversation", target_id=conv_id,
detail=f"将对话 {conv.title} 转换为知识图谱 {kb.name}",
)
return {
"code": 0,
"data": {
"knowledge_base_id": kb.id,
"knowledge_base_name": kb.name,
"graph": graph_data,
"knowledge_graph_count": 1,
"already_exists": False,
},
"message": f"知识图谱构建完成",
}
@router.get("/models")
def get_models(db: Session = Depends(get_db)):
models = db.query(AIModel).filter(AIModel.is_enabled == True).all()
result = []
for m in models:
provider = db.query(ModelProvider).filter(ModelProvider.id == m.provider_id).first()
result.append({
"id": m.id,
"name": m.name,
"display_name": getattr(m, "display_name", None) or m.name,
"provider_id": m.provider_id,
"provider_name": provider.name if provider else "",
"capabilities": m.capabilities or [],
"status": m.status,
"price_per_million_tokens": float(m.price_per_million_tokens or 0),
"is_enabled": m.is_enabled,
"is_default": bool(m.is_default) if hasattr(m, 'is_default') else False,
"total_calls": m.total_calls or 0,
"total_tokens": m.total_tokens or 0,
"created_at": str(m.created_at) if m.created_at else None,
})
result.sort(key=lambda x: (0 if x["is_default"] else 1, x["name"]))
return {"code": 0, "data": result, "message": "ok"}
Reference in new issue View Git Blame Copy Permalink