459 lines
20 KiB
Python
459 lines
20 KiB
Python
# -*- coding:utf-8 -*-
|
||
"""P2-③ 类内共性提取引擎：原子化 → 归一化 → 能力域 → 覆盖率。
|
||
|
||
流程（对单个 cluster 按需执行，用户/agent 触发，不做全批自动跑——成本控制）：
|
||
1. 原子化：类内文档按 20/批 LLM 提取原子需求（并发 Semaphore 3，purpose=utility），
|
||
落 opp_demand_atoms（snap×atom 一行）。
|
||
2. 归一化：原子文本向量化（rag embed 共享通道）→ 簇内 kNN + τ2(默认0.85) 并查集
|
||
得候选组 → **LLM 裁决**组内是否同一标准需求（语义铁律：相似度只当触发器）→
|
||
标准需求命名 → opp_std_demands。
|
||
3. 能力域：LLM 把标准需求分组到能力域（每簇一次调用）→ opp_demand_domains + 回填 domain_id。
|
||
4. 覆盖率：doc_count = 标准需求覆盖的 distinct snap 数；coverage = doc_count/类内文档数；
|
||
tier：core≥0.6 / ext≥0.4 / personal（阈值走 params：opp_cov_core/opp_cov_ext）。
|
||
|
||
状态：opp_clusters.normalize_status（init/running/done/failed + error 落 naming_evidence 旁路字段
|
||
normalize_error——ALTER 补列）。幂等：重跑先清该簇旧 atoms/std/domains 再算。
|
||
"""
|
||
import asyncio
|
||
import json
|
||
import logging
|
||
|
||
from .opp_common import get_db, new_id, rows_to_dicts, get_param
|
||
from .opp_mining import _cos, _centroid, demand_vec_id
|
||
|
||
logger = logging.getLogger("pipeline.opp_normalize")
|
||
|
||
# 客户端硬超时（秒）：llm_call 的 _timeout 只透传给治理端点，客户端 aiohttp 无超时——
|
||
# 上游挂起时客户端会无限等（实测 atomize 卡 6 分钟 atoms=0）。wait_for 快速失败+重试。
|
||
LLM_CLIENT_TIMEOUT = 120
|
||
LLM_RETRIES = 2
|
||
|
||
|
||
async def _llm(prompt, ctx, timeout=LLM_CLIENT_TIMEOUT):
|
||
"""llm_call 客户端超时包装：wait_for + 指数退避重试，全失败抛 RuntimeError。"""
|
||
from pipeline_service.llm_bridge import llm_call
|
||
last = None
|
||
for a in range(LLM_RETRIES):
|
||
try:
|
||
return await asyncio.wait_for(
|
||
llm_call(prompt, purpose="utility", timeout=timeout,
|
||
org_id=ctx.get("org_id") or "0",
|
||
user_id=ctx.get("user_id") or "",
|
||
session_id=ctx.get("session_id") or ""),
|
||
timeout=timeout + 10)
|
||
except Exception as e:
|
||
last = e
|
||
if a < LLM_RETRIES - 1:
|
||
await asyncio.sleep(2 ** a)
|
||
raise RuntimeError("LLM 调用失败(%d次): %s" % (LLM_RETRIES, str(last)[:150]))
|
||
|
||
P = {
|
||
"opp_norm_atom_batch": "20", # 原子化每批文档数
|
||
"opp_norm_tau2": "0.85", # 归一化相似度触发阈值
|
||
"opp_norm_topk": "10", # 原子 kNN 邻居数
|
||
"opp_cov_core": "0.6", # 核心共性阈值
|
||
"opp_cov_ext": "0.4", # 扩展共性阈值
|
||
}
|
||
|
||
|
||
async def get_norm_params(sor):
|
||
cfg = {}
|
||
for k, dv in P.items():
|
||
cfg[k] = await get_param(sor, k, dv)
|
||
for ik in ("opp_norm_atom_batch", "opp_norm_topk"):
|
||
try:
|
||
cfg[ik] = int(cfg[ik])
|
||
except (TypeError, ValueError):
|
||
cfg[ik] = int(P[ik])
|
||
for fk in ("opp_norm_tau2", "opp_cov_core", "opp_cov_ext"):
|
||
try:
|
||
cfg[fk] = float(cfg[fk])
|
||
except (TypeError, ValueError):
|
||
cfg[fk] = float(P[fk])
|
||
return cfg
|
||
|
||
|
||
async def _set_norm_status(sor, cluster_id, status, err=""):
|
||
await sor.sqlExe(
|
||
"UPDATE opp_clusters SET normalize_status=${s}$, normalize_error=${e}$ WHERE id=${c}$",
|
||
{"s": status, "e": err[:500], "c": cluster_id})
|
||
await sor.sqlExe("COMMIT", {})
|
||
|
||
|
||
async def _load_cluster(sor, cluster_id):
|
||
recs = await sor.sqlExe(
|
||
"SELECT c.*, b.org_id AS b_org FROM opp_clusters c "
|
||
"LEFT JOIN opp_mining_batches b ON b.id=c.batch_id WHERE c.id=${c}$",
|
||
{"c": cluster_id})
|
||
await sor.sqlExe("COMMIT", {})
|
||
if not recs:
|
||
return None
|
||
return rows_to_dicts(recs, limit=1)[0]
|
||
|
||
|
||
async def _cluster_snaps(sor, cluster_id):
|
||
recs = await sor.sqlExe(
|
||
"SELECT id, title, source, src_id FROM opp_demand_snap WHERE cluster_id=${c}$",
|
||
{"c": cluster_id})
|
||
await sor.sqlExe("COMMIT", {})
|
||
return rows_to_dicts(recs, limit=5000)
|
||
|
||
|
||
# ══════════════════ 1. 原子化 ══════════════════
|
||
|
||
async def _atomize(sor, snaps, cfg, ctx):
|
||
"""LLM 批量提取原子需求。返回 [(snap_id, atom_text)]。"""
|
||
from pipeline_service.llm_bridge import llm_call
|
||
sem = asyncio.Semaphore(3)
|
||
bs = cfg["opp_norm_atom_batch"]
|
||
results = []
|
||
lock = asyncio.Lock()
|
||
|
||
async def _one(chunk):
|
||
titles = "\n".join("%d. %s" % (i + 1, s["title"]) for i, s in enumerate(chunk))
|
||
prompt = (
|
||
"以下是同一类软件众包需求的标题（编号列表）。从每条标题提取它表达的**原子软件需求**"
|
||
"（如「微信小程序开发」「考勤记录」「合同到期提醒」），一条标题可提取1-3个原子需求，"
|
||
"无法判断的跳过。\n" + titles +
|
||
"\n\n只输出 JSON 数组，元素为 {\"no\": 编号, \"atom\": \"原子需求文本\"}，"
|
||
"不要解释、不要 markdown 代码块。")
|
||
async with sem:
|
||
try:
|
||
raw = await _llm(prompt, ctx, timeout=90)
|
||
raw = (raw or "").strip()
|
||
if raw.startswith("```"):
|
||
raw = raw.strip("`")
|
||
if raw.lower().startswith("json"):
|
||
raw = raw[4:]
|
||
arr = json.loads(raw)
|
||
items = []
|
||
for it in arr if isinstance(arr, list) else []:
|
||
if not isinstance(it, dict):
|
||
continue
|
||
no = it.get("no")
|
||
atom = str(it.get("atom") or "").strip()[:255]
|
||
# 容错：LLM 可能输出 "1"/1.0/True，统一 int() 转换后范围校验
|
||
try:
|
||
no_i = int(no)
|
||
except (TypeError, ValueError):
|
||
continue
|
||
if 1 <= no_i <= len(chunk) and atom:
|
||
items.append((chunk[no_i - 1]["id"], atom))
|
||
async with lock:
|
||
results.extend(items)
|
||
except Exception as e:
|
||
logger.debug("atomize chunk failed: %s", e)
|
||
async with lock:
|
||
results.extend([(s["id"], s["title"][:255]) for s in chunk])
|
||
|
||
chunks = [snaps[i:i + bs] for i in range(0, len(snaps), bs)]
|
||
await asyncio.gather(*(_one(ch) for ch in chunks))
|
||
# 产出率兜底：原子数 < 文档数 50% 时，缺原子的文档用标题作原子
|
||
# （众包标题本身即浓缩需求语义；保证覆盖率统计可算，2026-09-12 实测 46 文档仅 6 原子）
|
||
covered = set(sid for sid, _ in results)
|
||
if len(covered) < len(snaps) * 0.5:
|
||
for s in snaps:
|
||
if s["id"] not in covered:
|
||
results.append((s["id"], s["title"][:255]))
|
||
return results
|
||
|
||
|
||
# ══════════════════ 2. 归一化（embedding + LLM 裁决）══════════════════
|
||
|
||
async def _normalize(sor, atoms, project_id, cfg, ctx):
|
||
"""atoms=[(snap_id, atom_text)] → 标准需求组 [(name, [snap_ids], [atom_texts])]。"""
|
||
if not atoms:
|
||
return []
|
||
texts = [a[1] for a in atoms]
|
||
from pipeline_service import rag_client as rc
|
||
vecs, err = await rc.rag_embed_texts(project_id, texts, batch_size=10)
|
||
if err or not vecs or len(vecs) != len(texts):
|
||
raise RuntimeError("原子向量化失败: %s" % (err or "数量不符"))
|
||
|
||
# 簇内 kNN（VDB 临时 collection）
|
||
base = await _vdb_base(sor)
|
||
col = "opp_norm_%s" % new_id()[:12]
|
||
ok, d = await _vdb_post(sor, base, "/v1/createcollection", {
|
||
"colname": col, "fields": [
|
||
{"name": "id", "type": "str", "is_primary": True, "max_length": 64},
|
||
{"name": "vector", "type": "fvector", "dim": 1024},
|
||
{"name": "text", "type": "str", "max_length": 500}],
|
||
"description": "opp normalize workset", "metric": "COSINE"})
|
||
if not ok and "exist" not in str(d).lower():
|
||
raise RuntimeError("建归一化工作集失败: %s" % d)
|
||
rows = [{"id": "a%d" % i, "vector": vecs[i], "text": texts[i][:490]}
|
||
for i in range(len(texts))]
|
||
for i in range(0, len(rows), 500):
|
||
ok, d = await _vdb_post(sor, base, "/v1/upsert",
|
||
{"colname": col, "data": rows[i:i + 500]})
|
||
if not ok:
|
||
await _vdb_post(sor, base, "/v1/dropcollection", {"colname": col})
|
||
raise RuntimeError("归一化 upsert 失败: %s" % d)
|
||
nbr = {}
|
||
for i in range(len(texts)):
|
||
ok, d = await _vdb_post(sor, base, "/v1/query", {
|
||
"colname": col, "vector": vecs[i], "pagerows": cfg["opp_norm_topk"],
|
||
"output_fields": ["id"]})
|
||
if not ok:
|
||
await _vdb_post(sor, base, "/v1/dropcollection", {"colname": col})
|
||
raise RuntimeError("归一化 kNN 失败: %s" % d)
|
||
rr = (d.get("data") or {}).get("rows") or []
|
||
nbr["a%d" % i] = [(float(r.get("score", 0) or 0), r.get("id"))
|
||
for r in rr if r.get("id") and r.get("id") != "a%d" % i]
|
||
await _vdb_post(sor, base, "/v1/dropcollection", {"colname": col})
|
||
|
||
# τ2 并查集得候选组
|
||
ids = ["a%d" % i for i in range(len(texts))]
|
||
groups = _union_find_groups(ids, nbr, cfg["opp_norm_tau2"])
|
||
|
||
# LLM 裁决 + 命名（每组一次，并发 3）
|
||
from pipeline_service.llm_bridge import llm_call
|
||
sem = asyncio.Semaphore(3)
|
||
out = []
|
||
lock = asyncio.Lock()
|
||
|
||
async def _judge(g):
|
||
members = [int(x[1:]) for x in g]
|
||
uniq = []
|
||
seen = set()
|
||
for m in members:
|
||
if texts[m] not in seen:
|
||
seen.add(texts[m])
|
||
uniq.append(texts[m])
|
||
if len(uniq) == 1:
|
||
name, keep = uniq[0], members
|
||
else:
|
||
prompt = (
|
||
"以下是候选的同类原子软件需求（去重后）。判断它们是否表达**同一个标准软件需求**"
|
||
"（语义相同/包含关系算同一；明显不同功能不算）。\n" +
|
||
"\n".join("- " + u for u in uniq[:12]) +
|
||
"\n\n若同一：输出 JSON {\"same\": true, \"name\": \"标准需求名(≤12字)\"}；"
|
||
"若不同：输出 {\"same\": false, \"split\": [[编号...], [编号...]]} 给出应拆开的分组"
|
||
"（编号从1起，对应上面列表顺序）。只输出 JSON。")
|
||
async with sem:
|
||
try:
|
||
raw = await _llm(prompt, ctx, timeout=90)
|
||
raw = (raw or "").strip().strip("`")
|
||
if raw.lower().startswith("json"):
|
||
raw = raw[4:]
|
||
d2 = json.loads(raw)
|
||
except Exception as e:
|
||
logger.debug("judge failed, keep group: %s", e)
|
||
d2 = {"same": True, "name": uniq[0][:40]}
|
||
if d2.get("same"):
|
||
name = str(d2.get("name") or uniq[0]).strip()[:128]
|
||
keep = members
|
||
else:
|
||
# 拆分：每个子组单独成标准需求，命名用子组首条
|
||
async with lock:
|
||
for sub in (d2.get("split") or []):
|
||
idxs = [members[i - 1] for i in sub
|
||
if isinstance(i, int) and 1 <= i <= len(members)]
|
||
if idxs:
|
||
out.append((uniq[0][:128] if len(idxs) == 1 else
|
||
texts[idxs[0]][:128], idxs,
|
||
[texts[i] for i in idxs]))
|
||
return
|
||
async with lock:
|
||
out.append((name, keep, [texts[m] for m in keep]))
|
||
|
||
await asyncio.gather(*(_judge(g) for g in groups))
|
||
return out
|
||
|
||
|
||
def _union_find_groups(ids, nbr, tau):
|
||
parent = {v: v for v in ids}
|
||
|
||
def find(x):
|
||
while parent[x] != x:
|
||
parent[x] = parent[parent[x]]
|
||
x = parent[x]
|
||
return x
|
||
|
||
for v in ids:
|
||
for sc, nid in nbr.get(v, []):
|
||
if sc >= tau and nid in parent:
|
||
ra, rb = find(v), find(nid)
|
||
if ra != rb:
|
||
parent[rb] = ra
|
||
comps = {}
|
||
for v in ids:
|
||
comps.setdefault(find(v), []).append(v)
|
||
return sorted(comps.values(), key=len, reverse=True)
|
||
|
||
|
||
async def _vdb_base(sor):
|
||
recs = await sor.sqlExe("SELECT baseurl FROM upapp WHERE id='rag-vdb'", {})
|
||
await sor.sqlExe("COMMIT", {})
|
||
return (recs[0].baseurl or "").rstrip("/") if recs else ""
|
||
|
||
|
||
async def _vdb_post(sor, base, path, payload, timeout=40):
|
||
from .opp_mining import vdb_post
|
||
return await vdb_post(sor, base, path, payload, timeout=timeout)
|
||
|
||
|
||
# ══════════════════ 3. 能力域 ══════════════════
|
||
|
||
async def _domains(sor, cluster_id, org_id, std_rows, ctx):
|
||
"""LLM 把标准需求分组到能力域。std_rows=[{id,name}]。返回 [(domain_name, [std_ids])]。"""
|
||
if not std_rows:
|
||
return []
|
||
from pipeline_service.llm_bridge import llm_call
|
||
names = "\n".join("%d. %s" % (i + 1, s["name"]) for i, s in enumerate(std_rows))
|
||
prompt = (
|
||
"以下是某类软件的标准需求清单。把它们归入 2-6 个**能力域**"
|
||
"（如「人员管理」「工时与考勤」「合同管理」「报表统计」）。\n" + names +
|
||
"\n\n只输出 JSON 数组 [{\"domain\": \"域名\", \"nos\": [编号...]}]，覆盖全部编号，不解释。")
|
||
try:
|
||
raw = await _llm(prompt, ctx, timeout=90)
|
||
raw = (raw or "").strip().strip("`")
|
||
if raw.lower().startswith("json"):
|
||
raw = raw[4:]
|
||
arr = json.loads(raw)
|
||
except Exception as e:
|
||
logger.debug("domains failed: %s", e)
|
||
return [("通用能力", [s["id"] for s in std_rows])]
|
||
out = []
|
||
for g in arr if isinstance(arr, list) else []:
|
||
nos = g.get("nos") or []
|
||
ids = [std_rows[i - 1]["id"] for i in nos
|
||
if isinstance(i, int) and 1 <= i <= len(std_rows)]
|
||
if ids:
|
||
out.append((str(g.get("domain") or "未命名域")[:128], ids))
|
||
covered = set(i for _, ids in out for i in ids)
|
||
left = [s["id"] for s in std_rows if s["id"] not in covered]
|
||
if left:
|
||
out.append(("其他能力", left))
|
||
return out
|
||
|
||
|
||
# ══════════════════ 主流程 ══════════════════
|
||
|
||
async def normalize_cluster(sor, cluster_id, ctx):
|
||
"""对单个类别跑 原子化→归一化→能力域→覆盖率。返回 (ok, msg)。"""
|
||
cl = await _load_cluster(sor, cluster_id)
|
||
if not cl:
|
||
return False, "类别不存在"
|
||
org_id = str(cl.get("org_id") or cl.get("b_org") or "")
|
||
if org_id and org_id != (ctx.get("org_id") or ""):
|
||
return False, "类别不属于当前机构"
|
||
cfg = await get_norm_params(sor)
|
||
snaps = await _cluster_snaps(sor, cluster_id)
|
||
if not snaps:
|
||
return False, "类别内无需求快照"
|
||
project_id = ctx.get("project_id") or ""
|
||
|
||
await _set_norm_status(sor, cluster_id, "running")
|
||
try:
|
||
# 幂等：清旧结果
|
||
for tbl in ("opp_demand_atoms", "opp_std_demands", "opp_demand_domains"):
|
||
await sor.sqlExe("DELETE FROM %s WHERE cluster_id=${c}$" % tbl,
|
||
{"c": cluster_id})
|
||
await sor.sqlExe("COMMIT", {})
|
||
|
||
# 1 原子化
|
||
atoms = await _atomize(sor, snaps, cfg, ctx)
|
||
if not atoms:
|
||
await _set_norm_status(sor, cluster_id, "failed", "原子化无结果(LLM全失败)")
|
||
return False, "原子化无结果"
|
||
for sid, atom in atoms:
|
||
await sor.C("opp_demand_atoms", {
|
||
"id": new_id(), "cluster_id": cluster_id, "org_id": org_id,
|
||
"snap_id": sid, "atom_text": atom})
|
||
await sor.sqlExe("COMMIT", {})
|
||
|
||
# 2 归一化
|
||
groups = await _normalize(sor, atoms, project_id, cfg, ctx)
|
||
snap_by_id = {s["id"]: s for s in snaps}
|
||
std_rows = []
|
||
for name, members, atom_texts in groups:
|
||
sid_ = new_id()
|
||
snap_ids = sorted({atoms[m][0] for m in members})
|
||
await sor.C("opp_std_demands", {
|
||
"id": sid_, "cluster_id": cluster_id, "org_id": org_id,
|
||
"name": name[:128], "doc_count": len(snap_ids),
|
||
"coverage": 0.0, "tier": "personal",
|
||
"evidence_json": json.dumps(
|
||
{"atoms": atom_texts[:10], "snaps": snap_ids[:50]}, ensure_ascii=False)})
|
||
std_rows.append({"id": sid_, "name": name, "snap_ids": snap_ids})
|
||
# 回填原子的 std 归属
|
||
for m in members:
|
||
await sor.sqlExe(
|
||
"UPDATE opp_demand_atoms SET std_demand_id=${s}$ "
|
||
"WHERE cluster_id=${c}$ AND snap_id=${sn}$ AND atom_text=${a}$",
|
||
{"s": sid_, "c": cluster_id, "sn": atoms[m][0], "a": atoms[m][1]})
|
||
await sor.sqlExe("COMMIT", {})
|
||
|
||
# 4 覆盖率（先算再写 tier）
|
||
n_docs = len(snaps)
|
||
for s in std_rows:
|
||
cov = round(len(s["snap_ids"]) / n_docs, 4) if n_docs else 0.0
|
||
tier = "core" if cov >= cfg["opp_cov_core"] else (
|
||
"ext" if cov >= cfg["opp_cov_ext"] else "personal")
|
||
await sor.sqlExe(
|
||
"UPDATE opp_std_demands SET coverage=${v}$, tier=${t}$ WHERE id=${i}$",
|
||
{"v": cov, "t": tier, "i": s["id"]})
|
||
await sor.sqlExe("COMMIT", {})
|
||
|
||
# 3 能力域
|
||
doms = await _domains(sor, cluster_id, org_id,
|
||
[{"id": s["id"], "name": s["name"]} for s in std_rows], ctx)
|
||
for dname, std_ids in doms:
|
||
did = new_id()
|
||
await sor.C("opp_demand_domains", {
|
||
"id": did, "cluster_id": cluster_id, "org_id": org_id,
|
||
"parent_id": "", "name": dname, "std_count": len(std_ids)})
|
||
for i in range(0, len(std_ids), 200):
|
||
chunk = std_ids[i:i + 200]
|
||
ph = ",".join("${d%d}$" % j for j in range(len(chunk)))
|
||
args = {"did": did}
|
||
for j, sv in enumerate(chunk):
|
||
args["d%d" % j] = sv
|
||
await sor.sqlExe(
|
||
"UPDATE opp_std_demands SET domain_id=${did}$ WHERE id IN (%s)" % ph, args)
|
||
await sor.sqlExe("COMMIT", {})
|
||
|
||
await _set_norm_status(sor, cluster_id, "done")
|
||
core = sum(1 for s in std_rows
|
||
if len(s["snap_ids"]) / n_docs >= cfg["opp_cov_core"]) if n_docs else 0
|
||
return True, "归一化完成：原子 %d 条 → 标准需求 %d 个（核心 %d）→ 能力域 %d 个" % (
|
||
len(atoms), len(std_rows), core, len(doms))
|
||
except Exception as e:
|
||
logger.exception("normalize cluster %s failed", cluster_id)
|
||
await _set_norm_status(sor, cluster_id, "failed", str(e))
|
||
return False, "归一化失败: %s" % str(e)[:300]
|
||
|
||
|
||
async def coverage_report(sor, ctx, cluster_id):
|
||
"""覆盖率报告（org 隔离）。返回 (ok, dict|err)。"""
|
||
cl = await _load_cluster(sor, cluster_id)
|
||
if not cl:
|
||
return False, "类别不存在"
|
||
org_id = str(cl.get("org_id") or cl.get("b_org") or "")
|
||
if org_id and org_id != (ctx.get("org_id") or ""):
|
||
return False, "类别不属于当前机构"
|
||
recs = await sor.sqlExe(
|
||
"SELECT id, name, doc_count, coverage, tier, domain_id FROM opp_std_demands "
|
||
"WHERE cluster_id=${c}$ ORDER BY coverage DESC", {"c": cluster_id})
|
||
await sor.sqlExe("COMMIT", {})
|
||
stds = rows_to_dicts(recs, limit=200)
|
||
drecs = await sor.sqlExe(
|
||
"SELECT id, name, std_count FROM opp_demand_domains WHERE cluster_id=${c}$",
|
||
{"c": cluster_id})
|
||
await sor.sqlExe("COMMIT", {})
|
||
doms = rows_to_dicts(drecs, limit=50)
|
||
dom_name = {d["id"]: d["name"] for d in doms}
|
||
for s in stds:
|
||
s["domain"] = dom_name.get(s.get("domain_id") or "", "")
|
||
tiers = {}
|
||
for s in stds:
|
||
tiers[s["tier"]] = tiers.get(s["tier"], 0) + 1
|
||
return True, {
|
||
"cluster": {"id": cluster_id, "name": cl.get("name"),
|
||
"doc_count": cl.get("doc_count"),
|
||
"normalize_status": cl.get("normalize_status"),
|
||
"normalize_error": cl.get("normalize_error") or ""},
|
||
"tiers": tiers, "std_demands": stds, "domains": doms,
|
||
}
|