diff --git a/platform/backend/app/api/assistant.py b/platform/backend/app/api/assistant.py index 121e8a0..056c568 100644 --- a/platform/backend/app/api/assistant.py +++ b/platform/backend/app/api/assistant.py @@ -1,8 +1,11 @@ +import re +import json from fastapi import APIRouter, Depends, HTTPException from pydantic import BaseModel -from typing import List, Optional +from typing import List, Optional, Dict, Any from ..core.nvidia_client import call_llm from .auth import get_current_user +from .assistant_actions import ACTION_REGISTRY router = APIRouter(prefix="/api/assistant", tags=["assistant"]) @@ -17,45 +20,293 @@ PAGE_CONTEXTS = { "admin": "系统管理页面：管理用户、LLM配置、系统配置、采集类别、信息源、组织、查看运行日志等", } +# 可用操作清单（用于系统提示词） +from .assistant_actions import ACTION_REGISTRY +ACTION_DESCRIPTIONS = "\n".join([f"- {name}: {info['description']}" for name, info in ACTION_REGISTRY.items()]) + class ChatRequest(BaseModel): message: str page: str = "dashboard" history: List[dict] = [] +def _execute_action(action_name: str, params: Dict[str, Any], current_user) -> Dict[str, Any]: + """执行一个具体的操作""" + if action_name not in ACTION_REGISTRY: + return {"error": f"未知操作: {action_name}"} + func = ACTION_REGISTRY[action_name]["func"] + try: + result = func(**params) + return {"success": True, "result": result} + except Exception as e: + import traceback + return {"error": str(e), "traceback": traceback.format_exc()} + @router.post("/chat") def chat(request: ChatRequest, current_user=Depends(get_current_user)): + user_message = request.message.strip() + + # 1. 意图识别（规则优先）：用户问题涉及系统数据，直接调用工具，避免 LLM 漏掉 ACTION + intent, params = _detect_intent(user_message) + if intent: + result = _execute_action(intent, params, current_user) + reply = _format_action_result(intent, params, result) + return {"reply": reply, "actions": [{"action": intent, "params": params, "result": result}]} + + # 2. 走 LLM 对话（让 LLM 生成 ACTION） try: from ..database import SessionLocal - from ..models import SystemConfig + from ..models import SystemConfig, User db = SessionLocal() sc = db.query(SystemConfig).filter(SystemConfig.key == "assistant_system_prompt").first() base_prompt = sc.value if sc else None + user_rec = db.query(User).filter(User.id == getattr(current_user, 'id', None)).first() + username = user_rec.username if user_rec else (getattr(current_user, 'username', '用户')) db.close() except Exception: base_prompt = None + username = getattr(current_user, 'username', '用户') page_name = request.page page_guide = PAGE_CONTEXTS.get(page_name, "未知页面") - username = current_user.username if hasattr(current_user, 'username') else '用户' - system_prompt = base_prompt or f"""你是一个智能内容创作平台助手，帮助用户解答平台使用问题。 + system_prompt = base_prompt or f"""你是一个智能内容创作平台助手，能够通过调用工具与系统实时交互，为用户提供准确的数据和操作。 当前用户 {username} 正在查看：{page_guide} -你可以帮助用户： -1. 解释各个页面的功能和操作方法 -2. 解答平台使用中的疑问 -3. 提供操作建议和最佳实践 -4. 指导用户完成具体的操作步骤 +**重要规则**：对于任何涉及系统数据（选题、任务、统计、配置等）的问题，你必须使用提供的工具查询最新数据，不要依赖你的训练数据或猜测。 -请用友好、简洁的中文回答。如果不确定，建议用户参考相关页面或联系管理员。一次回答控制在200字以内。""" +可用工具： +{ACTION_DESCRIPTIONS} +**调用格式（必须严格遵守）**： +当需要执行工具时，在你的回复中包含单独一行（不要在行内有其他文字）： +**ACTION: action_name** {{"param1": "value1", ...}} + +示例对话： +用户：系统现在有多少选题？ +助手：稍等，我为你查询一下。 + +**ACTION: get_system_stats** {{}} + +用户：我想看最近5条任务日志 +助手：好的。 + +**ACTION: get_recent_task_logs** {{"limit":5}} + +工具执行后，系统会返回结果，你再根据结果生成最终回复（如「当前选题总数是42个」）。 + +注意： +- 必须严格使用上述格式，工具名和参数必须匹配 +- 参数必须是合法JSON，缺失必要参数时应询问用户 +- 如果用户问题模糊，先澄清再调用工具 +- 回复内容简洁友好，不超过200字 +""" messages = request.history[-20:] conversation = "\n".join([f"{'用户' if m.get('role') == 'user' else '助手'}: {m.get('content', '')}" for m in messages]) prompt_text = f"{conversation}\n用户: {request.message}\n助手:" try: - reply = call_llm(prompt_text, system_prompt=system_prompt, temperature=0.5, max_tokens=1000) - return {"reply": reply} + reply = call_llm(prompt_text, system_prompt=system_prompt, temperature=0.5, max_tokens=1500) + + # 检测是否有操作指令（支持多行） + action_pattern = re.compile(r'\*\*ACTION:\s*(\w+)\*\*\s*(\{.*\})', re.DOTALL) + action_matches = list(action_pattern.finditer(reply)) + action_results = [] + processed_reply = reply + + for match in action_matches: + action_name = match.group(1).strip() + args_str = match.group(2).strip() + try: + args = json.loads(args_str) + # 执行操作 + exec_result = _execute_action(action_name, args, current_user) + action_results.append({ + "action": action_name, + "params": args, + "result": exec_result + }) + # 从回复中移除 ACTION 标记行 + processed_reply = processed_reply.replace(match.group(0), '').strip() + except json.JSONDecodeError: + action_results.append({"action": action_name, "error": "参数JSON格式错误"}) + + # 如果有执行结果，追加简要说明到回复中 + if action_results: + summary_lines = [] + for ar in action_results: + if "error" in ar["result"]: + summary_lines.append(f"❌ 操作 {ar['action']} 失败: {ar['result']['error']}") + else: + summary_lines.append(f"✅ 操作 {ar['action']} 已完成。") + processed_reply = processed_reply + "\n\n" + "\n".join(summary_lines) + + return {"reply": processed_reply.strip(), "actions": action_results} except Exception as e: return {"reply": f"抱歉，AI助手暂时无法响应，请稍后重试。"} + + +def _detect_intent(message: str) -> (str, Dict[str, Any]): + """基于关键词的意图识别，返回 (action_name, params) 或 (None, {{}})""" + msg = message.lower() + + # 查询系统统计（选题总数、各状态数量、案例数） + if any(k in msg for k in ['统计', '总数', '多少选题', '选题数量', '系统数据', '数据总览']): + return "get_system_stats", {} + + # 查看选题列表 + if any(k in msg for k in ['选题列表', '查看选题', '选题有哪些', '列出选题', '选题']): + # 默认返回第一页20条，可后续细化 + page = 1 + page_size = 10 + # 检测是否指定状态 + status = None + if any(k in msg for k in ['待处理', '待审核', '待审查', '待发布', '已发布']): + status_map = {'待处理': 'pending', '待审查': 'review', '待发布': 'ready', '已发布': 'published'} + for k, v in status_map.items(): + if k in msg: + status = v + break + page_size = 20 if '全部' in msg else page_size + return "list_topics", {"page": page, "page_size": page_size, "status": status, "field": None} + + # 查询特定选题 + if any(k in msg for k in ['查看选题', '选题详情', '选题详情', '获取选题', '选题id']): + # 尝试提取ID（大写字母+数字 或 LIVING/TECH/M0604 等格式） + import re + match = re.search(r'[A-Z0-9\-]{4,}', msg) + if match: + topic_id = match.group(0) + return "get_topic", {"topic_id": topic_id} + return None, {} + + # 最近任务日志 + if any(k in msg for k in ['任务日志', '执行记录', '任务记录', '最近任务']): + limit = 5 if '最近' in msg or '查看' in msg else 10 + # 检查是否指定模块 + module_id = None + for mod in ['collect', 'generate', 'optimize', 'metrics', 'trends']: + if mod in msg: + module_id = f"scheduled_{mod}" + break + return "get_task_status", {"module_id": module_id} if module_id else {"module_id": None} + + # 手动触发任务 + if any(k in msg for k in ['触发', '运行', '执行', '采集', '生成', '优化', '指标']): + module_map = { + 'collect': 'scheduled_collect', + '采集': 'scheduled_collect', + 'generate': 'scheduled_generate', + '生成': 'scheduled_generate', + 'optimize': 'scheduled_optimize', + '优化': 'scheduled_optimize', + 'optimize_sources': 'scheduled_optimize_sources', + 'source': 'scheduled_optimize_sources', + 'metrics': 'scheduled_metrics_sync', + '指标': 'scheduled_metrics_sync', + 'trend': 'scheduled_trends', + '趋势': 'scheduled_trends' + } + for key, val in module_map.items(): + if key in msg: + return "trigger_task", {"module_id": val} + # 如果无法识别具体模块，不触发 + return None, {} + + # 搜索 + if any(k in msg for k in ['搜索', '查询', '查找']): + # 提取搜索词（简单：把问题其余部分作为查询） + # 这里较难简单提取，暂不自动触发，留给 LLM + return None, {} + + # 默认不触发 + return None, {} + + +def _format_action_result(action_name: str, params: Dict[str, Any], result: Dict[str, Any]) -> str: + """将工具执行结果格式化为自然语言回复""" + if "error" in result: + return f"⚠️ 执行 {action_name} 失败：{result['error']}" + + res = result.get("result", result) + + if action_name == "get_system_stats": + t = res.get("topics", {}) + return (f"📊 系统统计：\n" + f"• 选题总数：{t.get('total', 0)}\n" + f"• 待处理：{t.get('pending', 0)}\n" + f"• 待审查：{t.get('review', 0)}\n" + f"• 草稿：{t.get('draft', 0)}\n" + f"• 待发布：{t.get('ready', 0)}\n" + f"• 已发布：{t.get('published', 0)}\n" + f"• 本周新增：{t.get('new_this_week', 0)}\n" + f"• 案例总数：{res.get('cases', 0)}") + + if action_name == "list_topics": + items = res.get("items", []) + total = res.get("total", 0) + page = res.get("page", 1) + if not items: + return f"当前没有符合条件的选题（总计 {total} 条）。" + lines = [f"📋 选题列表（第 {page} 页，显示 {len(items)} 条，共 {total} 条）："] + for it in items: + lines.append(f" • {it['id']}：{it['title'][:30]}... [{it['status']}]") + return "\n".join(lines) + + if action_name == "get_topic": + if "error" in res: + return f"❌ {res['error']}" + t = res + return (f"📄 选题详情 [{t['id']}]：\n" + f"标题：{t['title']}\n" + f"领域：{t['field']}\n" + f"状态：{t['status']}\n" + f"优先级：{t['priority']} (分：{t['priority_score']})") + + if action_name == "get_recent_task_logs": + logs = res[:5] if isinstance(res, list) else [] + if not logs: + return "暂无任务日志。" + lines = ["📝 最近任务记录："] + for l in logs: + status_icon = {"success":"✅","failed":"❌","running":"🔄"}.get(l['status'],"⬜") + time_str = l['started_at'][:10] if l.get('started_at') else "?" + lines.append(f" {status_icon} {l['module_id']} - {l['status']} ({time_str})") + return "\n".join(lines) + + if action_name == "trigger_task": + if "error" in res: + return f"❌ 触发失败：{res['error']}" + return f"⏳ 已触发任务：{params.get('module_id')}，正在后台执行。" + + if action_name == "search_web": + if "error" in res: + return f"❌ 搜索失败：{res['error']}" + results = res.get("results", []) + count = res.get("count", len(results)) + if not results: + return "未找到相关搜索结果。" + lines = [f"🔍 搜索到 {count} 条结果："] + for i, r in enumerate(results[:3], 1): + lines.append(f" {i}. {r.get('title','无标题')}\n {r.get('url','')}") + if count > 3: + lines.append(f" ... 还有 {count-3} 条") + return "\n".join(lines) + + if action_name == "get_task_status": + logs = res[:5] if isinstance(res, list) else [] + if not logs: + return "未找到任务日志。" + lines = ["📋 任务状态："] + for l in logs: + status_icon = {"success":"✅","failed":"❌","running":"🔄"}.get(l['status'],"⬜") + time_str = l['started_at'][:10] if l.get('started_at') else "?" + msg = (l.get('message') or '') + msg_snippet = (msg[:30] + '...') if len(msg) > 30 else msg[:30] + lines.append(f" {status_icon} {l['module_id']} - {l['status']} ({time_str})") + if msg_snippet: + lines.append(f" {msg_snippet}") + return "\n".join(lines) + + # 默认返回原始结果 + return f"✅ 操作 {action_name} 完成。" diff --git a/platform/backend/app/api/assistant_actions.py b/platform/backend/app/api/assistant_actions.py new file mode 100644 index 0000000..c02b9f5 --- /dev/null +++ b/platform/backend/app/api/assistant_actions.py @@ -0,0 +1,248 @@ +""" +AI Assistant 可执行操作注册表 +""" +import json +from datetime import datetime, timezone, timedelta +from typing import Dict, Any, List, Optional + +from ..database import SessionLocal +from ..models import Topic, ContentTask, TaskLog, SystemConfig, TopicField, Case +from ..core.scheduler import scheduler as bg_scheduler + +# -------------------- 工具函数 -------------------- + +def list_topics( + page: int = 1, + page_size: int = 20, + status: Optional[str] = None, + field: Optional[str] = None +) -> Dict[str, Any]: + """获取选题列表，支持分页和筛选""" + db = SessionLocal() + try: + query = db.query(Topic) + if status: + query = query.filter(Topic.status == status) + if field: + query = query.filter(Topic.field_name == field) + total = query.count() + items = query.order_by(Topic.priority.desc(), Topic.created_at.desc()) \ + .offset((page - 1) * page_size) \ + .limit(page_size) \ + .all() + result_items = [] + for t in items: + result_items.append({ + "id": t.id, + "title": t.title, + "field": t.field_name, + "status": t.status, + "priority": t.priority, + "priority_score": t.priority_score, + "created_at": t.created_at.isoformat() if t.created_at else None + }) + return { + "total": total, + "page": page, + "page_size": page_size, + "items": result_items + } + finally: + db.close() + + +def get_topic(topic_id: str) -> Dict[str, Any]: + """获取单个选题详情""" + db = SessionLocal() + try: + t = db.query(Topic).filter(Topic.id == topic_id).first() + if not t: + return {"error": f"选题 {topic_id} 不存在"} + return { + "id": t.id, + "title": t.title, + "field": t.field_name, + "status": t.status, + "priority": t.priority, + "priority_score": t.priority_score, + "core_concept": t.core_concept, + "audience_pain": t.audience_pain, + "unique_angle": t.unique_angle, + "cases": t.cases or [], + "tags": t.tags or [], + "platform_urls": t.platform_urls or {}, + "created_at": t.created_at.isoformat() if t.created_at else None + } + finally: + db.close() + + +def get_recent_task_logs(limit: int = 10) -> List[Dict[str, Any]]: + """获取最近的任务执行日志""" + db = SessionLocal() + try: + logs = db.query(TaskLog).order_by(TaskLog.started_at.desc()).limit(limit).all() + return [{ + "id": l.id, + "module_id": l.module_id, + "status": l.status, + "message": l.message, + "started_at": l.started_at.isoformat() if l.started_at else None, + "finished_at": l.finished_at.isoformat() if l.finished_at else None, + "triggered_by": l.triggered_by + } for l in logs] + finally: + db.close() + + +def get_system_stats() -> Dict[str, Any]: + """获取系统概览统计数据""" + db = SessionLocal() + try: + total_topics = db.query(Topic).count() + pending = db.query(Topic).filter(Topic.status == "pending").count() + review = db.query(Topic).filter(Topic.status == "review").count() + draft = db.query(Topic).filter(Topic.status == "draft").count() + ready = db.query(Topic).filter(Topic.status == "ready").count() + published = db.query(Topic).filter(Topic.status == "published").count() + total_cases = db.query(Case).count() + week_ago = datetime.now(timezone.utc).replace(hour=0, minute=0, second=0, microsecond=0) - timedelta(days=7) + new_this_week = db.query(Topic).filter(Topic.created_at >= week_ago).count() + return { + "topics": { + "total": total_topics, + "pending": pending, + "review": review, + "draft": draft, + "ready": ready, + "published": published, + "new_this_week": new_this_week + }, + "cases": total_cases, + "date": datetime.now(timezone.utc).isoformat() + } + finally: + db.close() + + +def get_task_status(module_id: Optional[str] = None) -> List[Dict[str, Any]]: + """查询任务日志（如 collector、generator、optimizer 等）""" + db = SessionLocal() + try: + query = db.query(TaskLog) + if module_id: + query = query.filter(TaskLog.module_id == module_id) + logs = query.order_by(TaskLog.started_at.desc()).limit(10).all() + return [{ + "id": l.id, + "module_id": l.module_id, + "status": l.status, + "message": l.message, + "started_at": l.started_at.isoformat() if l.started_at else None, + "finished_at": l.finished_at.isoformat() if l.finished_at else None, + "triggered_by": l.triggered_by + } for l in logs] + finally: + db.close() + + +def trigger_task(module_id: str) -> Dict[str, Any]: + """手动触发定时任务（如 collector、generator、optimizer、metrics_sync 等）""" + # 只允许触发特定任务 + allowed = ["scheduled_collect", "scheduled_generate", "scheduled_optimize", + "scheduled_optimize_sources", "scheduled_metrics_sync", "scheduled_trends"] + if module_id not in allowed: + return {"error": f"不允许手动触发该任务: {module_id}"} + job_id = module_id if module_id.startswith("scheduled_") else f"scheduled_{module_id}" + # 检查任务是否存在 + job = bg_scheduler.get_job(job_id) + if not job: + return {"error": f"任务不存在: {job_id}"} + try: + # 如果任务正在运行，返回提示 + # APScheduler 的 run_job 会立即执行 + bg_scheduler.run_job(job_id) + return {"success": True, "message": f"任务 {job_id} 已触发执行"} + except Exception as e: + return {"error": str(e)} + + +def search_web(query: str, max_results: int = 5) -> List[Dict[str, Any]]: + """联网搜索（通过本项目的 search_utils）""" + try: + from ...scripts.search_utils import search as web_search + results = web_search(query, max_results) + return {"success": True, "results": results, "count": len(results)} + except Exception as e: + return {"error": str(e)} + + +def list_system_configs() -> List[Dict[str, Any]]: + """列出系统配置项""" + db = SessionLocal() + try: + configs = db.query(SystemConfig).all() + return [{"key": c.key, "value": c.value, "description": c.description} for c in configs] + finally: + db.close() + + +# Action registry +ACTION_REGISTRY = { + "list_topics": { + "func": list_topics, + "description": "获取选题列表，支持分页(page, page_size)和筛选(status, field)", + "params_schema": { + "page": {"type": "integer", "default": 1, "desc": "页码"}, + "page_size": {"type": "integer", "default": 20, "desc": "每页条数"}, + "status": {"type": "string", "enum": ["pending", "review", "draft", "ready", "published"], "desc": "按状态筛选"}, + "field": {"type": "string", "desc": "按领域筛选"} + } + }, + "get_topic": { + "func": get_topic, + "description": "获取单个选题详情", + "params_schema": { + "topic_id": {"type": "string", "desc": "选题ID，如 A01 或 LIVING-001-26"} + } + }, + "get_recent_task_logs": { + "func": get_recent_task_logs, + "description": "获取最近的任务执行日志", + "params_schema": { + "limit": {"type": "integer", "default": 10, "desc": "返回条数"} + } + }, + "get_system_stats": { + "func": get_system_stats, + "description": "获取系统概览统计（选题总数、各状态数量、案例数等）", + "params_schema": {} + }, + "get_task_status": { + "func": get_task_status, + "description": "查询任务日志，可按模块ID筛选", + "params_schema": { + "module_id": {"type": "string", "desc": "可选：筛选特定模块ID（如 scheduled_collect）"} + } + }, + "trigger_task": { + "func": trigger_task, + "description": "手动触发定时任务（采集、生成、优化、指标同步等）", + "params_schema": { + "module_id": {"type": "string", "desc": "任务模块ID：scheduled_collect/scheduled_generate/scheduled_optimize/scheduled_optimize_sources/scheduled_metrics_sync/scheduled_trends"} + } + }, + "search_web": { + "func": search_web, + "description": "执行网络搜索，返回结构化结果", + "params_schema": { + "query": {"type": "string", "desc": "搜索关键词"}, + "max_results": {"type": "integer", "default": 5, "desc": "返回结果数量"} + } + }, + "list_system_configs": { + "func": list_system_configs, + "description": "列出所有系统配置项（key/value）", + "params_schema": {} + } +}