1507 lines
59 KiB
Python
1507 lines
59 KiB
Python
"""
|
||
增强版 Chat API — Phase 2.5
|
||
包含: Skill ID统一(MarketApp映射), 真实Skill执行, Agent工作流注入,
|
||
多轮上下文保持, BM25 KB检索, Function Calling循环增强
|
||
"""
|
||
from fastapi import APIRouter, Depends, HTTPException, Query
|
||
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,
|
||
gen_uuid
|
||
)
|
||
from ..auth import get_current_user
|
||
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
|
||
|
||
# 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("&", "&").replace("<", "<").replace(">", ">")
|
||
.replace('"', """).replace("'", "'"))
|
||
|
||
|
||
def _outbound_proxy() -> str:
|
||
"""出网代理地址。容器内直连 apihub.agnes-ai.com 会被对端重置
|
||
(httpx RemoteProtocolError: Server disconnected without sending a response)，
|
||
必须经 172.18.0.1:7890 智能路由代理(其对国内 AGNES 域名走直连)。"""
|
||
import os
|
||
return os.environ.get("HTTPS_PROXY") or os.environ.get("HTTP_PROXY") or "http://172.18.0.1:7890"
|
||
|
||
|
||
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。"
|
||
payload = {
|
||
"model": ai_model.name,
|
||
"prompt": prompt,
|
||
"size": "1024x768",
|
||
"extra_body": {"response_format": "url"},
|
||
}
|
||
try:
|
||
async with httpx.AsyncClient(timeout=timeout, proxy=_outbound_proxy()) as client:
|
||
resp = await client.post(api_url, headers=headers, json=payload)
|
||
if resp.status_code != 200:
|
||
return f"⚠️ 图像生成失败 (HTTP {resp.status_code}): {resp.text[:200]}"
|
||
data = resp.json()
|
||
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)}')
|
||
except Exception as e:
|
||
return f"⚠️ 图像生成异常：{str(e)[:200]}"
|
||
|
||
# ============ 视频生成 (异步任务 + 轮询) ============
|
||
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。"
|
||
base = api_url.split("/v1")[0] if "/v1" in api_url else api_url
|
||
payload = {
|
||
"model": ai_model.name,
|
||
"prompt": prompt,
|
||
"height": 768,
|
||
"width": 1152,
|
||
"num_frames": 121,
|
||
"frame_rate": 24,
|
||
}
|
||
try:
|
||
async with httpx.AsyncClient(timeout=timeout, proxy=_outbound_proxy()) as client:
|
||
resp = await client.post(api_url, headers=headers, json=payload)
|
||
if resp.status_code != 200:
|
||
return f"⚠️ 视频生成失败 (HTTP {resp.status_code}): {resp.text[:200]}"
|
||
data = resp.json()
|
||
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)
|
||
r2 = await client.get(f"{base}/v1/videos/{video_id}", headers=headers)
|
||
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 '生成超时，请稍后在对话中重试'}"
|
||
except Exception as e:
|
||
return f"⚠️ 视频生成异常：{str(e)[:200]}"
|
||
|
||
return "⚠️ 未知的媒体模型类型。"
|
||
|
||
|
||
|
||
# ============================================================
|
||
# 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
|
||
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 "",
|
||
})
|
||
|
||
return results
|
||
|
||
|
||
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
|
||
"""
|
||
if not kb_ids:
|
||
return ""
|
||
|
||
kbs = db.query(KnowledgeBase).filter(KnowledgeBase.id.in_(kb_ids)).all()
|
||
if not kbs:
|
||
return ""
|
||
|
||
# 收集所有文档段落
|
||
# 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])
|
||
return f"参考知识库: {kb_names}"
|
||
|
||
# BM25检索最相关的段落
|
||
top_segments = _bm25_search(query, all_segments, top_k=8)
|
||
|
||
# 拼接上下文
|
||
context_parts = [seg["content"] for seg in top_segments]
|
||
context = "\n".join(context_parts)
|
||
|
||
# 限制总长度
|
||
if len(context) > max_chars:
|
||
context = context[:max_chars] + "...(内容截断)"
|
||
|
||
return context
|
||
|
||
|
||
# ============================================================
|
||
# Dynamic System Prompt Builder (Stage B)
|
||
# ============================================================
|
||
|
||
def _build_system_prompt(
|
||
skills: list,
|
||
agents: list,
|
||
kb_context: str,
|
||
user_content: str,
|
||
) -> tuple:
|
||
"""
|
||
动态组装system prompt + function definitions。
|
||
返回: (system_prompt_str, function_definitions_list)
|
||
"""
|
||
system_parts = ["你是AIFlow平台的AI助手，具备专业的知识储备和工具调用能力。"]
|
||
|
||
function_definitions = []
|
||
|
||
# Stage B.1: Agent角色注入
|
||
if agents:
|
||
for agent in agents:
|
||
role = agent.get("role_prompt", "")
|
||
instructions = agent.get("instructions", "")
|
||
if role:
|
||
system_parts.append(f"当前角色设定: {role}")
|
||
if instructions:
|
||
system_parts.append(f"角色行为指引: {instructions}")
|
||
|
||
# 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}")
|
||
system_parts.append("重要规则: 当用户请求涉及上述技能工具的处理范围时, 你必须优先通过function calling调用对应工具, 而不是直接回复文字。调用后将获得工具执行结果, 基于结果生成最终回复。")
|
||
|
||
# 为每个skill构建function definition
|
||
for s in skills:
|
||
# 从skill_content提取指令注入system prompt
|
||
instructions = s.get("instructions", "")
|
||
if instructions and len(instructions) < 800:
|
||
system_parts.append(f"技能「{s['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}")
|
||
|
||
system_prompt = "\n".join(system_parts)
|
||
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
|
||
|
||
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 [],
|
||
"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_use_tools = req.use_tools or False
|
||
|
||
if not effective_skill_ids and not effective_agent_ids and not effective_kb_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", [])
|
||
if effective_skill_ids or effective_agent_ids:
|
||
effective_use_tools = True
|
||
|
||
# 保存用户消息(含skill/agent/kb选择用于上下文保持)
|
||
user_msg = Message(
|
||
id=gen_uuid(),
|
||
conversation_id=conv_id,
|
||
role="user",
|
||
content=req.content,
|
||
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,
|
||
created_at=cst_now(),
|
||
)
|
||
db.add(user_msg)
|
||
|
||
# ===== 核心聊天引擎 =====
|
||
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
|
||
# 解析模型：支持 Auto 模式（model_id='auto' 或未指定时，按能力自动匹配平台默认模型）
|
||
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 []
|
||
if ai_model and ("image" in caps or "video" in caps):
|
||
try:
|
||
ai_content = await _generate_media(db, ai_model, req.content, current_user.id)
|
||
except Exception as e:
|
||
ai_content = f"⚠️ 媒体生成失败：{str(e)[:200]}"
|
||
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)
|
||
|
||
# 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,
|
||
)
|
||
|
||
# 构建消息列表
|
||
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})
|
||
|
||
messages.append({"role": "user", "content": req.content})
|
||
|
||
# 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) 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}
|
||
|
||
tool_result = _execute_skill_tool(
|
||
db, func_name, func_args, skills, req.content
|
||
)
|
||
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模型进行对话。"
|
||
else:
|
||
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)
|
||
|
||
# 更新used_skills
|
||
used_skills = []
|
||
if skills:
|
||
used_skills = [s["name"] for s in skills]
|
||
ai_msg.used_skills = used_skills
|
||
|
||
# 更新对话统计
|
||
conv.knowledge_graph_count = (conv.knowledge_graph_count or 0) + 1
|
||
conv.updated_at = cst_now()
|
||
|
||
# 记录Token使用
|
||
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)
|
||
|
||
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,
|
||
},
|
||
},
|
||
"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"}
|
||
|