311 lines
16 KiB
Python
311 lines
16 KiB
Python
"""AI 辅助创建（小白友好）：大白话 → 结构化草稿 + 资源冲突检测。
|
||
|
||
两个能力：
|
||
1. draft_from_blab：用户用自己的话描述「我想要一个干什么的 Agent / 技能 / 模板」，
|
||
由当前默认模型解析成结构化草稿（名称/说明/规则/指令/提示模板字段），
|
||
用户在表单里确认修改后再正式创建——AI 只出草稿，不直接落库。
|
||
2. conflict_scan：创建/修改前的冲突检测，回答「Agent、技能、模板之间会不会打架」：
|
||
- 重名/高相似名称
|
||
- 同名约束互相矛盾（规则 vs 技能 vs 模板约束）
|
||
- 关键词互斥（如「口语化」对「书面语」「输出 JSON」对「段落正文」）
|
||
|
||
冲突不阻断创建，返回 warning 列表由前端展示，用户判断后可继续。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import re
|
||
from difflib import SequenceMatcher
|
||
from sqlmodel import Session, select
|
||
|
||
from .llm import chat_json
|
||
from .models import Agent, AgentSkill, AgentPromptTemplate
|
||
|
||
# 语义互斥对：同一条消息里同时要求两边就会打架。
|
||
MUTUAL_EXCLUSIVE_PAIRS: list[tuple[str, str]] = [
|
||
("口语", "书面"), ("大白话", "书面语"), ("口语", "公文"),
|
||
("简短", "详细"), ("精炼", "展开"), ("长文", "短句"),
|
||
("严肃", "幽默"), ("专业", "接地气"), ("正式", "轻松"),
|
||
("列表", "段落"), ("分点", "一段到底"), ("json", "口语"),
|
||
("markdown", "纯文本"), ("提问", "陈述"), ("第一人称", "第三人称"),
|
||
]
|
||
|
||
DRAFT_SYSTEM_PROMPT = """你是私域Agent系统的配置助手。用户不懂 AI，只会用自己的话描述想要的东西。
|
||
你的任务是把用户的描述整理成结构化草稿，输出 JSON。
|
||
|
||
输出要求：
|
||
- 只输出一个合法 JSON 对象，不要 markdown 代码块标记，不要解释。
|
||
- 名称要具体（4-12 个字），说明是一句完整的话。
|
||
- 用户没提到的字段用空字符串或空列表，不要编造细节。
|
||
- 规则和指令用大白话写，像跟同事交代事情，不写 AI 术语。
|
||
|
||
三种资源的 JSON 结构：
|
||
1. agent（智能体）:
|
||
{"kind": "agent", "name": "...", "role": "production|qc|composer|service", "desc": "...",
|
||
"rules": ["规则1", "规则2"], "content_tags": ["标签"], "skill_suggestions": ["想要的能力描述"],
|
||
"advice": "给用户的一句提醒或建议"}
|
||
2. skill（技能）:
|
||
{"kind": "skill", "name": "...", "desc": "...",
|
||
"instructions": ["步骤1", "步骤2"], "advice": "..."}
|
||
3. template（提示模板）:
|
||
{"kind": "template", "name": "...", "desc": "...",
|
||
"role_prompt": "你是...（角色设定）", "task_template": "...",
|
||
"constraints": ["约束1"], "style": "文风描述", "advice": "..."}
|
||
|
||
role 判断：写内容的选 production，检查内容质量的选 qc，把多段合成完整稿的选 composer，
|
||
客服对话选 service。不确定时选 production。"""
|
||
|
||
|
||
|
||
# ---------------- 知识库检索注入 ----------------
|
||
|
||
def _kb_ids_for_kind(session: Session, kind: str) -> list[int]:
|
||
"""按草稿类型挑选参与检索的知识库：
|
||
agent/skill/template 都查全部业务库——草稿可能涉及任何板块；
|
||
检索相关性排序会自动挑最相关的块。"""
|
||
from .models import KnowledgeBase
|
||
names = {"心法", "技法", "功法", "情欲内容", "产品信息", "销售逻辑"}
|
||
out = []
|
||
for kb in session.exec(select(KnowledgeBase)).all():
|
||
if kb.name in names:
|
||
out.append(kb.id)
|
||
return out
|
||
|
||
|
||
def retrieve_kb_context(session: Session, description: str, kind: str, k: int = 8) -> tuple[list[dict], str]:
|
||
"""按描述检索知识块，返回 (hits, 注入 prompt 的上下文文本)。检索失败返回空不阻塞。"""
|
||
try:
|
||
from .rag import pipeline as rag_pipeline
|
||
kb_ids = _kb_ids_for_kind(session, kind)
|
||
if not kb_ids:
|
||
return [], ""
|
||
hits = rag_pipeline.search_kbs_remote(
|
||
session, kb_ids, description, k=k,
|
||
user_id=0, organization_id=1)
|
||
hits = [h for h in hits if str(h.get("text") or "").strip()][:k]
|
||
if not hits:
|
||
return [], ""
|
||
parts = []
|
||
for i, h in enumerate(hits, 1):
|
||
kb_name = str(h.get("kb_name") or "")
|
||
src = str((h.get("meta") or {}).get("file") or "")
|
||
text = re.sub(r"\s+", " ", str(h.get("text") or ""))[:420]
|
||
parts.append(f"[知识{i}]（{kb_name}·{src}）{text}")
|
||
return hits, "\n\n".join(parts)
|
||
except Exception as exc: # 检索失败不阻塞草稿生成
|
||
print(f"[ai-assist] 知识库检索失败（不影响生成）: {exc}", flush=True)
|
||
return [], ""
|
||
|
||
|
||
def _pick_model(session: Session, organization_id: int | None = None) -> tuple[int, str] | None:
|
||
"""挑一个可用的 Provider/模型：优先全局默认，其次任意启用的。"""
|
||
from .models import ModelProvider, ModelCatalog, AppSetting
|
||
setting = session.exec(select(AppSetting).where(AppSetting.key == "model_default")).first()
|
||
if setting:
|
||
try:
|
||
data = __import__("json").loads(setting.value)
|
||
provider = session.get(ModelProvider, int(data.get("provider_id") or 0))
|
||
if provider and provider.enabled and (organization_id is None or int(provider.organization_id or 1) in {0, int(organization_id)} ) and data.get("model_name"):
|
||
return provider.id, str(data["model_name"])
|
||
except (ValueError, TypeError, KeyError):
|
||
pass
|
||
# Do not fall back to the first enabled built-in provider: it may have no
|
||
# development credential while a user-configured provider is usable.
|
||
from .secret_store import resolve_secret
|
||
providers = session.exec(select(ModelProvider).where(ModelProvider.enabled == True)).all() # noqa: E712
|
||
if organization_id is not None:
|
||
providers = [p for p in providers if int(p.organization_id or 1) in {0, int(organization_id)}]
|
||
for provider in providers:
|
||
if provider.default_model and resolve_secret(provider.api_key, f"provider-{provider.id}"):
|
||
return provider.id, provider.default_model
|
||
return None
|
||
|
||
|
||
def draft_from_blab(session: Session, kind: str, description: str,
|
||
user_id: int = 0, organization_id: int = 1,
|
||
advice: str = "", previous: dict | None = None) -> dict:
|
||
"""大白话 → 结构化草稿。无可用模型时抛 ValueError（前端显示可读提示）。
|
||
advice+previous：人工未确认时的"按建议重新生成"通道——意见最高优先级注入。"""
|
||
description = str(description or "").strip()
|
||
if not description:
|
||
raise ValueError("请先用一段话描述你想要的东西")
|
||
if len(description) < 5:
|
||
raise ValueError("描述太短了，多说几句你想要它干什么、风格是什么")
|
||
if kind not in {"agent", "skill", "template"}:
|
||
raise ValueError("不支持的资源类型")
|
||
model = _pick_model(session)
|
||
if not model:
|
||
raise ValueError("还没有可用的模型，请先到「模型管理」添加 Provider")
|
||
provider_id, model_name = model
|
||
hits, kb_context = retrieve_kb_context(session, description, kind)
|
||
user_msg = f"类型：{kind}\n用户说：{description[:2000]}"
|
||
if advice:
|
||
user_msg += f"\n\n【人工修改意见（最高优先级，必须逐条落实，输出修改后的完整草稿）】\n{advice[:1500]}"
|
||
if previous:
|
||
import json as _json
|
||
user_msg += "\n\n【上一版草稿（在其基础上按意见修改）】\n" + _json.dumps(previous, ensure_ascii=False)[:2500]
|
||
if kb_context:
|
||
user_msg += ("\n\n【系统知识库相关内容（供你理解业务语境，草稿的规则/指令应贴合这些真实知识）】\n"
|
||
+ kb_context)
|
||
result = chat_json(
|
||
session,
|
||
[{"role": "system", "content": DRAFT_SYSTEM_PROMPT},
|
||
{"role": "user", "content": user_msg}],
|
||
provider_id, model_name,
|
||
agent_name="AI 辅助创建", user_id=user_id, organization_id=organization_id,
|
||
max_retries=1, temperature=0.4,
|
||
)
|
||
if result.get("status") != "ok" or not isinstance(result.get("json"), dict):
|
||
raise ValueError(f"AI 没能理解这段描述：{str(result.get('error'))[:160]}")
|
||
draft = result["json"]
|
||
draft["kind"] = kind
|
||
# 模型信息回传，前端展示这次草稿花了多少
|
||
draft["_meta"] = {"provider": result.get("provider"), "model": result.get("model"),
|
||
"tokens": result.get("tokens"), "duration_ms": result.get("duration_ms"),
|
||
"kb_hits": [{"kb_id": h.get("kb_id"), "kb_name": h.get("kb_name"),
|
||
"file": str((h.get("meta") or {}).get("file") or ""),
|
||
"excerpt": re.sub(r"\s+", " ", str(h.get("text") or ""))[:80]}
|
||
for h in hits]}
|
||
return draft
|
||
|
||
|
||
# ---------------- 冲突检测 ----------------
|
||
|
||
def _norm(text: str) -> str:
|
||
return re.sub(r"[\s，。、！？：；「」\-—_]", "", str(text or "")).lower()
|
||
|
||
|
||
def _similar(a: str, b: str) -> float:
|
||
return SequenceMatcher(None, _norm(a), _norm(b)).ratio()
|
||
|
||
|
||
def _find_mutex_clauses(text: str, known_conflicts: list[str]) -> list[str]:
|
||
"""在一段文本里找互斥表述。返回「A ↔ B」列表。"""
|
||
found: list[str] = []
|
||
norm = _norm(text)
|
||
for left, right in MUTUAL_EXCLUSIVE_PAIRS:
|
||
l, r = _norm(left), _norm(right)
|
||
if l in norm and r in norm:
|
||
pair = f"{left} ↔ {right}"
|
||
if pair not in known_conflicts:
|
||
found.append(pair)
|
||
return found
|
||
|
||
|
||
def _single_side_hits(text: str) -> set[str]:
|
||
"""文本里出现的互斥词（任一侧），用于跨资源单侧冲突比对。"""
|
||
norm = _norm(text)
|
||
hits: set[str] = set()
|
||
for left, right in MUTUAL_EXCLUSIVE_PAIRS:
|
||
if _norm(left) in norm:
|
||
hits.add(_norm(left))
|
||
if _norm(right) in norm:
|
||
hits.add(_norm(right))
|
||
return hits
|
||
|
||
|
||
def conflict_scan(session: Session, kind: str, payload: dict,
|
||
exclude_id: int = 0) -> dict:
|
||
"""扫描与现有资源的冲突。返回 {conflicts: [...], warnings: [...]}。
|
||
|
||
conflicts 是应该处理的（重名），warnings 是提醒用户判断的（相似名/互斥表述）。
|
||
"""
|
||
conflicts: list[str] = []
|
||
warnings: list[str] = []
|
||
name = str(payload.get("name") or "").strip()
|
||
norm_name = _norm(name)
|
||
if not norm_name:
|
||
return {"conflicts": ["名称不能为空"], "warnings": []}
|
||
|
||
# 1. 同类重名 → conflict
|
||
if kind == "agent":
|
||
rows = [(a.id, a.name, a) for a in session.exec(select(Agent)).all()]
|
||
elif kind == "skill":
|
||
rows = [(sk.id, sk.name, sk) for sk in session.exec(select(AgentSkill)).all()]
|
||
else:
|
||
rows = [(t.id, t.name, t) for t in session.exec(select(AgentPromptTemplate)).all()]
|
||
for row_id, row_name, _row in rows:
|
||
if row_id == exclude_id:
|
||
continue
|
||
if _norm(row_name) == norm_name:
|
||
conflicts.append(f"已存在同名的{'Agent' if kind == 'agent' else '技能' if kind == 'skill' else '提示模板'}「{row_name}」，换一个名字")
|
||
continue
|
||
if _similar(name, row_name) >= 0.82:
|
||
warnings.append(f"与现有「{row_name}」名字很像（相似度 {int(_similar(name, row_name) * 100)}%），确认不是重复创建？")
|
||
|
||
# 2. 互斥表述：收集本资源的所有文字
|
||
own_texts: list[str] = [name, str(payload.get("desc") or "")]
|
||
if kind == "agent":
|
||
own_texts += [str(item) for item in (payload.get("rules") or [])]
|
||
elif kind == "skill":
|
||
own_texts += [str(item) for item in (payload.get("instructions") or [])]
|
||
else:
|
||
own_texts += [str(payload.get("role_prompt") or ""), str(payload.get("task_template") or ""),
|
||
str(payload.get("style") or "")]
|
||
own_texts += [str(item) for item in (payload.get("constraints") or [])]
|
||
own_conflicts: list[str] = []
|
||
for text in own_texts:
|
||
for pair in _find_mutex_clauses(text, own_conflicts):
|
||
own_conflicts.append(pair)
|
||
if own_conflicts:
|
||
warnings.append("这段描述里有互相打架的要求：" + "；".join(f"「{pair}」" for pair in own_conflicts)
|
||
+ "，AI 不知道该听哪个")
|
||
|
||
# 3. Agent 与挂接技能的冲突（创建/编辑 Agent 时）
|
||
if kind == "agent":
|
||
skill_ids = [int(item) for item in (payload.get("skill_ids") or []) if str(item).isdigit()]
|
||
for skill_id in skill_ids:
|
||
skill = session.get(AgentSkill, skill_id)
|
||
if not skill:
|
||
continue
|
||
skill_text = skill.instructions
|
||
own_pairs: set[tuple[str, str]] = set()
|
||
for text in own_texts:
|
||
own_pairs.update(_find_mutex_clauses(text, []))
|
||
skill_pairs = set(_find_mutex_clauses(skill_text, []))
|
||
# 双侧都在（两边各自出现完整互斥对）→ 直接冲突
|
||
for l, r in own_pairs:
|
||
for l2, r2 in skill_pairs:
|
||
if _norm(l) in (_norm(l2), _norm(r2)) or _norm(r) in (_norm(l2), _norm(r2)):
|
||
warnings.append(f"与挂接的技能「{skill.name}」要求相反：{l} ↔ {r}，生成时会互相拉扯")
|
||
break
|
||
# 单侧词冲突：本资源只说「口语」，技能只说「书面」→ 两边拉扯
|
||
own_sides: set[str] = set()
|
||
for text in own_texts:
|
||
own_sides.update(_single_side_hits(text))
|
||
skill_sides = _single_side_hits(skill_text)
|
||
for l, r in MUTUAL_EXCLUSIVE_PAIRS:
|
||
if _norm(l) in own_sides and _norm(r) in skill_sides:
|
||
warnings.append(f"与挂接的技能「{skill.name}」要求相反：{l} ↔ {r}，生成时会互相拉扯")
|
||
elif _norm(r) in own_sides and _norm(l) in skill_sides:
|
||
warnings.append(f"与挂接的技能「{skill.name}」要求相反：{r} ↔ {l}，生成时会互相拉扯")
|
||
|
||
# 4. Agent 与其提示模板的约束冲突
|
||
if kind == "agent":
|
||
template_id = payload.get("prompt_template_id")
|
||
if template_id and str(template_id).isdigit():
|
||
template = session.get(AgentPromptTemplate, int(template_id))
|
||
if template:
|
||
template_texts = [template.role_prompt, template.task_template, template.style] + list(template.constraints or [])
|
||
template_sides: set[str] = set()
|
||
for text in template_texts:
|
||
template_sides.update(_single_side_hits(text))
|
||
own_sides_t: set[str] = set()
|
||
for text in own_texts:
|
||
own_sides_t.update(_single_side_hits(text))
|
||
for l, r in MUTUAL_EXCLUSIVE_PAIRS:
|
||
if _norm(l) in own_sides_t and _norm(r) in template_sides:
|
||
warnings.append(f"与提示模板「{template.name}」的约束相反：{l} ↔ {r}，模板会跟规则打架")
|
||
elif _norm(r) in own_sides_t and _norm(l) in template_sides:
|
||
warnings.append(f"与提示模板「{template.name}」的约束相反：{r} ↔ {l}，模板会跟规则打架")
|
||
|
||
return {"conflicts": conflicts, "warnings": warnings}
|
||
|
||
|
||
def conflict_scan_text(session: Session, kind: str, payload: dict, exclude_id: int = 0) -> dict:
|
||
"""conflict_scan 的文本摘要版，供 LLM 草稿后立即回检。"""
|
||
return conflict_scan(session, kind, payload, exclude_id)
|
||
|
||
|
||
__all__ = ["draft_from_blab", "conflict_scan", "conflict_scan_text",
|
||
"MUTUAL_EXCLUSIVE_PAIRS", "DRAFT_SYSTEM_PROMPT"]
|