feat(normalize): P2-③类内共性提取引擎——opp_normalize.py(原子化LLM合批并发→embedding+τ2并查集候选组→LLM裁决归一(相似度只当触发器)→能力域分组→覆盖率三层core≥0.6/ext≥0.4/personal);3新表atoms/std_demands/domains+clusters补normalize_status/error列;2工具normalize_cluster/coverage_report(org隔离+未归一先报错引导);params种子5项
This commit is contained in:
parent
2bf137e650
commit
1e64ab7e75
@ -38,6 +38,11 @@
|
||||
{"name": "opp_mine_big_split", "value": "150", "desc": "超大簇质心二次细分阈值(防模板标题过度聚合)"},
|
||||
{"name": "opp_mine_keep_batches", "value": "3", "desc": "每机构保留最近N个批次工作集collection,旧批自动drop"},
|
||||
{"name": "opp_embed_batch", "value": "10", "desc": "rag /embed 单批文本上限"},
|
||||
{"name": "opp_emb_cache_col", "value": "opp_demand_emb_cache", "desc": "共享embedding缓存collection(全机构复用)"}
|
||||
{"name": "opp_emb_cache_col", "value": "opp_demand_emb_cache", "desc": "共享embedding缓存collection(全机构复用)"},
|
||||
{"name": "opp_norm_atom_batch", "value": "20", "desc": "原子化每批文档数(LLM合批)"},
|
||||
{"name": "opp_norm_tau2", "value": "0.85", "desc": "归一化相似度触发阈值(只当触发器,LLM裁决)"},
|
||||
{"name": "opp_norm_topk", "value": "10", "desc": "原子kNN邻居数"},
|
||||
{"name": "opp_cov_core", "value": "0.6", "desc": "核心共性覆盖率阈值"},
|
||||
{"name": "opp_cov_ext", "value": "0.4", "desc": "扩展共性覆盖率阈值"}
|
||||
]
|
||||
}
|
||||
|
||||
@ -75,6 +75,18 @@
|
||||
"type": "str",
|
||||
"length": 32
|
||||
},
|
||||
{
|
||||
"name": "normalize_status",
|
||||
"title": "归一化状态(init/running/done/failed)",
|
||||
"type": "str",
|
||||
"length": 16,
|
||||
"default": "init"
|
||||
},
|
||||
{
|
||||
"name": "normalize_error",
|
||||
"title": "归一化失败原因",
|
||||
"type": "text"
|
||||
},
|
||||
{
|
||||
"name": "created_at",
|
||||
"title": "创建时间",
|
||||
|
||||
61
models/opp_demand_atoms.json
Normal file
61
models/opp_demand_atoms.json
Normal file
@ -0,0 +1,61 @@
|
||||
{
|
||||
"summary": [
|
||||
{
|
||||
"name": "opp_demand_atoms",
|
||||
"title": "需求原子(类内LLM原子化结果)",
|
||||
"primary": [
|
||||
"id"
|
||||
],
|
||||
"catelog": "entity"
|
||||
}
|
||||
],
|
||||
"fields": [
|
||||
{
|
||||
"name": "id",
|
||||
"title": "主键ID",
|
||||
"type": "str",
|
||||
"length": 32,
|
||||
"nullable": "no"
|
||||
},
|
||||
{
|
||||
"name": "cluster_id",
|
||||
"title": "所属类别ID",
|
||||
"type": "str",
|
||||
"length": 32,
|
||||
"nullable": "no"
|
||||
},
|
||||
{
|
||||
"name": "org_id",
|
||||
"title": "机构ID",
|
||||
"type": "str",
|
||||
"length": 32,
|
||||
"nullable": "no"
|
||||
},
|
||||
{
|
||||
"name": "snap_id",
|
||||
"title": "来源需求快照ID",
|
||||
"type": "str",
|
||||
"length": 32,
|
||||
"nullable": "no"
|
||||
},
|
||||
{
|
||||
"name": "atom_text",
|
||||
"title": "原子需求文本(LLM提取)",
|
||||
"type": "str",
|
||||
"length": 255,
|
||||
"nullable": "no"
|
||||
},
|
||||
{
|
||||
"name": "std_demand_id",
|
||||
"title": "归一后标准需求ID",
|
||||
"type": "str",
|
||||
"length": 32
|
||||
},
|
||||
{
|
||||
"name": "created_at",
|
||||
"title": "创建时间",
|
||||
"type": "timestamp",
|
||||
"nullable": "no"
|
||||
}
|
||||
]
|
||||
}
|
||||
61
models/opp_demand_domains.json
Normal file
61
models/opp_demand_domains.json
Normal file
@ -0,0 +1,61 @@
|
||||
{
|
||||
"summary": [
|
||||
{
|
||||
"name": "opp_demand_domains",
|
||||
"title": "需求能力域(两层需求模型上层)",
|
||||
"primary": [
|
||||
"id"
|
||||
],
|
||||
"catelog": "entity"
|
||||
}
|
||||
],
|
||||
"fields": [
|
||||
{
|
||||
"name": "id",
|
||||
"title": "主键ID",
|
||||
"type": "str",
|
||||
"length": 32,
|
||||
"nullable": "no"
|
||||
},
|
||||
{
|
||||
"name": "cluster_id",
|
||||
"title": "所属类别ID",
|
||||
"type": "str",
|
||||
"length": 32,
|
||||
"nullable": "no"
|
||||
},
|
||||
{
|
||||
"name": "org_id",
|
||||
"title": "机构ID",
|
||||
"type": "str",
|
||||
"length": 32,
|
||||
"nullable": "no"
|
||||
},
|
||||
{
|
||||
"name": "parent_id",
|
||||
"title": "父能力域ID(空=顶层)",
|
||||
"type": "str",
|
||||
"length": 32
|
||||
},
|
||||
{
|
||||
"name": "name",
|
||||
"title": "能力域名(如:人员管理/工时与考勤)",
|
||||
"type": "str",
|
||||
"length": 128,
|
||||
"nullable": "no"
|
||||
},
|
||||
{
|
||||
"name": "std_count",
|
||||
"title": "下辖标准需求数",
|
||||
"type": "int",
|
||||
"nullable": "no",
|
||||
"default": "0"
|
||||
},
|
||||
{
|
||||
"name": "created_at",
|
||||
"title": "创建时间",
|
||||
"type": "timestamp",
|
||||
"nullable": "no"
|
||||
}
|
||||
]
|
||||
}
|
||||
81
models/opp_std_demands.json
Normal file
81
models/opp_std_demands.json
Normal file
@ -0,0 +1,81 @@
|
||||
{
|
||||
"summary": [
|
||||
{
|
||||
"name": "opp_std_demands",
|
||||
"title": "标准需求(归一化结果)",
|
||||
"primary": [
|
||||
"id"
|
||||
],
|
||||
"catelog": "entity"
|
||||
}
|
||||
],
|
||||
"fields": [
|
||||
{
|
||||
"name": "id",
|
||||
"title": "主键ID",
|
||||
"type": "str",
|
||||
"length": 32,
|
||||
"nullable": "no"
|
||||
},
|
||||
{
|
||||
"name": "cluster_id",
|
||||
"title": "所属类别ID",
|
||||
"type": "str",
|
||||
"length": 32,
|
||||
"nullable": "no"
|
||||
},
|
||||
{
|
||||
"name": "org_id",
|
||||
"title": "机构ID",
|
||||
"type": "str",
|
||||
"length": 32,
|
||||
"nullable": "no"
|
||||
},
|
||||
{
|
||||
"name": "name",
|
||||
"title": "标准需求名(LLM裁决命名)",
|
||||
"type": "str",
|
||||
"length": 128,
|
||||
"nullable": "no"
|
||||
},
|
||||
{
|
||||
"name": "domain_id",
|
||||
"title": "能力域ID(两层模型上层)",
|
||||
"type": "str",
|
||||
"length": 32
|
||||
},
|
||||
{
|
||||
"name": "doc_count",
|
||||
"title": "覆盖文档数",
|
||||
"type": "int",
|
||||
"nullable": "no",
|
||||
"default": "0"
|
||||
},
|
||||
{
|
||||
"name": "coverage",
|
||||
"title": "覆盖率(覆盖文档数/类内文档数)",
|
||||
"type": "double",
|
||||
"length": 6,
|
||||
"dec": 4,
|
||||
"default": "0"
|
||||
},
|
||||
{
|
||||
"name": "tier",
|
||||
"title": "层级(core核心/ext扩展/personal个性)",
|
||||
"type": "str",
|
||||
"length": 16,
|
||||
"default": "personal"
|
||||
},
|
||||
{
|
||||
"name": "evidence_json",
|
||||
"title": "证据(原子样例JSON)",
|
||||
"type": "text"
|
||||
},
|
||||
{
|
||||
"name": "created_at",
|
||||
"title": "创建时间",
|
||||
"type": "timestamp",
|
||||
"nullable": "no"
|
||||
}
|
||||
]
|
||||
}
|
||||
101
mysql.ddl.sql
101
mysql.ddl.sql
@ -36,6 +36,38 @@ comment '研发报告'
|
||||
;
|
||||
|
||||
|
||||
-- /home/ymq/work/repos/pipeline-opportunity/models/opp_demand_atoms.json
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
-- 建库时请用以下语句，支持emoji字符
|
||||
-- CREATE DATABASE mydb CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;
|
||||
drop table if exists opp_demand_atoms;
|
||||
CREATE TABLE opp_demand_atoms
|
||||
(
|
||||
|
||||
`id` VARCHAR(32) NOT NULL comment '主键ID',
|
||||
`cluster_id` VARCHAR(32) NOT NULL comment '所属类别ID',
|
||||
`org_id` VARCHAR(32) NOT NULL comment '机构ID',
|
||||
`snap_id` VARCHAR(32) NOT NULL comment '来源需求快照ID',
|
||||
`atom_text` VARCHAR(255) NOT NULL comment '原子需求文本(LLM提取)',
|
||||
`std_demand_id` VARCHAR(32) comment '归一后标准需求ID',
|
||||
`created_at` TIMESTAMP DEFAULT CURRENT_TIMESTAMP NOT NULL comment '创建时间'
|
||||
|
||||
|
||||
,primary key(id)
|
||||
|
||||
|
||||
)
|
||||
CHARACTER SET utf8mb4
|
||||
COLLATE utf8mb4_unicode_ci
|
||||
engine=innodb
|
||||
comment '需求原子(类内LLM原子化结果)'
|
||||
;
|
||||
|
||||
|
||||
-- /home/ymq/work/repos/pipeline-opportunity/models/opp_approvals.json
|
||||
|
||||
|
||||
@ -92,6 +124,8 @@ CREATE TABLE opp_clusters
|
||||
`heat_rank` int DEFAULT '0' comment '热度排名',
|
||||
`naming_evidence` longtext CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci comment '命名证据(样例标题JSON)',
|
||||
`centroid_snap_id` VARCHAR(32) comment '质心需求快照ID(代表样本)',
|
||||
`normalize_status` VARCHAR(16) DEFAULT 'init' comment '归一化状态(init/running/done/failed)',
|
||||
`normalize_error` longtext CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci comment '归一化失败原因',
|
||||
`created_at` TIMESTAMP DEFAULT CURRENT_TIMESTAMP NOT NULL comment '创建时间'
|
||||
|
||||
|
||||
@ -106,6 +140,73 @@ comment '需求类别(聚类结果)'
|
||||
;
|
||||
|
||||
|
||||
-- /home/ymq/work/repos/pipeline-opportunity/models/opp_demand_domains.json
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
-- 建库时请用以下语句，支持emoji字符
|
||||
-- CREATE DATABASE mydb CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;
|
||||
drop table if exists opp_demand_domains;
|
||||
CREATE TABLE opp_demand_domains
|
||||
(
|
||||
|
||||
`id` VARCHAR(32) NOT NULL comment '主键ID',
|
||||
`cluster_id` VARCHAR(32) NOT NULL comment '所属类别ID',
|
||||
`org_id` VARCHAR(32) NOT NULL comment '机构ID',
|
||||
`parent_id` VARCHAR(32) comment '父能力域ID(空=顶层)',
|
||||
`name` VARCHAR(128) NOT NULL comment '能力域名(如:人员管理/工时与考勤)',
|
||||
`std_count` int NOT NULL DEFAULT '0' comment '下辖标准需求数',
|
||||
`created_at` TIMESTAMP DEFAULT CURRENT_TIMESTAMP NOT NULL comment '创建时间'
|
||||
|
||||
|
||||
,primary key(id)
|
||||
|
||||
|
||||
)
|
||||
CHARACTER SET utf8mb4
|
||||
COLLATE utf8mb4_unicode_ci
|
||||
engine=innodb
|
||||
comment '需求能力域(两层需求模型上层)'
|
||||
;
|
||||
|
||||
|
||||
-- /home/ymq/work/repos/pipeline-opportunity/models/opp_std_demands.json
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
-- 建库时请用以下语句，支持emoji字符
|
||||
-- CREATE DATABASE mydb CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;
|
||||
drop table if exists opp_std_demands;
|
||||
CREATE TABLE opp_std_demands
|
||||
(
|
||||
|
||||
`id` VARCHAR(32) NOT NULL comment '主键ID',
|
||||
`cluster_id` VARCHAR(32) NOT NULL comment '所属类别ID',
|
||||
`org_id` VARCHAR(32) NOT NULL comment '机构ID',
|
||||
`name` VARCHAR(128) NOT NULL comment '标准需求名(LLM裁决命名)',
|
||||
`domain_id` VARCHAR(32) comment '能力域ID(两层模型上层)',
|
||||
`doc_count` int NOT NULL DEFAULT '0' comment '覆盖文档数',
|
||||
`coverage` double(6,4) DEFAULT '0' comment '覆盖率(覆盖文档数/类内文档数)',
|
||||
`tier` VARCHAR(16) DEFAULT 'personal' comment '层级(core核心/ext扩展/personal个性)',
|
||||
`evidence_json` longtext CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci comment '证据(原子样例JSON)',
|
||||
`created_at` TIMESTAMP DEFAULT CURRENT_TIMESTAMP NOT NULL comment '创建时间'
|
||||
|
||||
|
||||
,primary key(id)
|
||||
|
||||
|
||||
)
|
||||
CHARACTER SET utf8mb4
|
||||
COLLATE utf8mb4_unicode_ci
|
||||
engine=innodb
|
||||
comment '标准需求(归一化结果)'
|
||||
;
|
||||
|
||||
|
||||
-- /home/ymq/work/repos/pipeline-opportunity/models/opp_mining_batches.json
|
||||
|
||||
|
||||
|
||||
@ -89,6 +89,10 @@ OPP_TOOLS = [
|
||||
parameters={"batch_id": "批次ID"}, category="mining"),
|
||||
ToolDefinition(name="opp_cluster_detail", description="查看某类别的类内需求明细样例（每条带来源URL可查证），用于写可行性报告时引用证据",
|
||||
parameters={"cluster_id": "类别ID", "limit": "样例条数(默认20)"}, category="mining"),
|
||||
ToolDefinition(name="opp_normalize_cluster", description="对某类别做类内共性提取（后台执行）：原子化→语义归一→能力域→覆盖率统计。完成后用 opp_coverage_report 看核心/扩展/个性三层共性需求。耗时与类内文档数成正比（LLM 调用较多，只对要深挖的类别执行）",
|
||||
parameters={"cluster_id": "类别ID"}, category="mining"),
|
||||
ToolDefinition(name="opp_coverage_report", description="查看某类别的共性需求覆盖率报告：核心共性(覆盖率≥60%)/扩展共性(≥40%)/个性需求三层 + 能力域分组 + 每条标准需求的覆盖文档数与证据",
|
||||
parameters={"cluster_id": "类别ID"}, category="mining"),
|
||||
]
|
||||
|
||||
# ══════════════════ 产线 prompt 片段 ══════════════════
|
||||
@ -494,6 +498,40 @@ async def _h_cluster_detail(sor, p, ctx):
|
||||
"样例(带来源可查证)": samples}, ensure_ascii=False, default=str)
|
||||
|
||||
|
||||
async def _h_normalize_cluster(sor, p, ctx):
|
||||
from .opp_normalize import normalize_cluster
|
||||
cid = (p.get("cluster_id") or "").strip()
|
||||
if not cid:
|
||||
return _fmt(False, "cluster_id 必填")
|
||||
ok, msg = await normalize_cluster(sor, cid, ctx)
|
||||
return _fmt(ok, msg)
|
||||
|
||||
|
||||
async def _h_coverage_report(sor, p, ctx):
|
||||
from .opp_normalize import coverage_report
|
||||
cid = (p.get("cluster_id") or "").strip()
|
||||
if not cid:
|
||||
return _fmt(False, "cluster_id 必填")
|
||||
ok, res = await coverage_report(sor, ctx, cid)
|
||||
if not ok:
|
||||
return _fmt(False, res)
|
||||
cl = res["cluster"]
|
||||
if cl.get("normalize_status") != "done":
|
||||
return _fmt(False, "该类别尚未完成共性提取（status=%s%s），先调 opp_normalize_cluster" % (
|
||||
cl.get("normalize_status"), ("：" + cl.get("normalize_error")) if cl.get("normalize_error") else ""))
|
||||
tiers = res["tiers"]
|
||||
rows = [{"标准需求": s["name"], "能力域": s["domain"], "覆盖文档": s["doc_count"],
|
||||
"覆盖率": "%.1f%%" % (float(s["coverage"] or 0) * 100), "层级": s["tier"],
|
||||
"证据原子": (json.loads(s.get("evidence_json") or "{}").get("atoms") or [])[:3]}
|
||||
for s in res["std_demands"]]
|
||||
return json.dumps({
|
||||
"类别": cl["name"], "类内文档数": cl["doc_count"],
|
||||
"层级统计": {"核心(≥60%)": tiers.get("core", 0), "扩展(≥40%)": tiers.get("ext", 0),
|
||||
"个性": tiers.get("personal", 0)},
|
||||
"能力域": [{"域": d["name"], "标准需求数": d["std_count"]} for d in res["domains"]],
|
||||
"标准需求": rows}, ensure_ascii=False, default=str)
|
||||
|
||||
|
||||
OPP_HANDLERS = {
|
||||
"opp_hot_software": _h_hot_software,
|
||||
"opp_daily_ai_tenders": _h_daily_ai,
|
||||
@ -518,6 +556,9 @@ OPP_HANDLERS = {
|
||||
"opp_mining_status": _h_mining_status,
|
||||
"opp_list_clusters": _h_list_clusters,
|
||||
"opp_cluster_detail": _h_cluster_detail,
|
||||
# 类内共性提取（P2）
|
||||
"opp_normalize_cluster": _h_normalize_cluster,
|
||||
"opp_coverage_report": _h_coverage_report,
|
||||
}
|
||||
|
||||
|
||||
|
||||
427
pipeline_opportunity/opp_normalize.py
Normal file
427
pipeline_opportunity/opp_normalize.py
Normal file
@ -0,0 +1,427 @@
|
||||
# -*- 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")
|
||||
|
||||
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_call(prompt, purpose="utility", timeout=90,
|
||||
org_id=ctx.get("org_id") or "0",
|
||||
user_id=ctx.get("user_id") or "",
|
||||
session_id=ctx.get("session_id") or "")
|
||||
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 []:
|
||||
no = it.get("no")
|
||||
atom = str(it.get("atom") or "").strip()[:255]
|
||||
if isinstance(no, int) and 1 <= no <= len(chunk) and atom:
|
||||
items.append((chunk[no - 1]["id"], atom))
|
||||
async with lock:
|
||||
results.extend(items)
|
||||
except Exception as e:
|
||||
logger.debug("atomize chunk failed: %s", e)
|
||||
|
||||
chunks = [snaps[i:i + bs] for i in range(0, len(snaps), bs)]
|
||||
await asyncio.gather(*(_one(ch) for ch in chunks))
|
||||
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_call(prompt, purpose="utility", timeout=90,
|
||||
org_id=ctx.get("org_id") or "0",
|
||||
user_id=ctx.get("user_id") or "",
|
||||
session_id=ctx.get("session_id") or "")
|
||||
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_call(prompt, purpose="utility", timeout=90,
|
||||
org_id=ctx.get("org_id") or "0",
|
||||
user_id=ctx.get("user_id") or "",
|
||||
session_id=ctx.get("session_id") or "")
|
||||
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,
|
||||
}
|
||||
42
scripts/p2_create_tables.py
Normal file
42
scripts/p2_create_tables.py
Normal file
@ -0,0 +1,42 @@
|
||||
# -*- coding:utf-8 -*-
|
||||
"""P2-③ 建表：opp_demand_atoms / opp_std_demands / opp_demand_domains（幂等 IF NOT EXISTS）。"""
|
||||
import asyncio, os, re, sys
|
||||
W = "/d/pipeline/pipeline-app"; os.chdir(W); sys.path.insert(0, W)
|
||||
from appPublic.folderUtils import ProgramPath
|
||||
from appPublic.jsonConfig import getConfig
|
||||
from appPublic.event_dispatcher import EventDispatcher
|
||||
from sqlor.dbpools import DBPools
|
||||
from ahserver.serverenv import ServerEnv
|
||||
p = ProgramPath(); c = getConfig(W, {"workdir": W, "ProgramPath": p}); DBPools(c.databases)
|
||||
se = ServerEnv(); se.event_dispatcher = EventDispatcher(); se.get_module_dbname = lambda m: "pipeline"
|
||||
|
||||
NEW = ["opp_demand_atoms", "opp_std_demands", "opp_demand_domains"]
|
||||
|
||||
async def main():
|
||||
ddl = open("/tmp/p2_tables.sql", encoding="utf-8").read()
|
||||
async with DBPools().sqlorContext("pipeline") as sor:
|
||||
for stmt in [s.strip() for s in ddl.split(";") if s.strip()]:
|
||||
m = re.match(r"(?is)CREATE\s+TABLE\s+(?:IF\s+NOT\s+EXISTS\s+)?`?(\w+)`?", stmt)
|
||||
if not m or m.group(1) not in NEW:
|
||||
continue
|
||||
await sor.sqlExe(stmt, {})
|
||||
await sor.sqlExe("COMMIT", {})
|
||||
print("table OK:", m.group(1))
|
||||
for t in NEW:
|
||||
r = await sor.sqlExe(
|
||||
"SELECT COUNT(*) c FROM information_schema.TABLES WHERE table_schema=DATABASE() AND table_name=${t}$", {"t": t})
|
||||
await sor.sqlExe("COMMIT", {})
|
||||
print("verify", t, "exists:", (r[0].c if r else 0))
|
||||
# 索引
|
||||
for name, tbl, cols in [("idx_opp_atoms_cluster", "opp_demand_atoms", "cluster_id"),
|
||||
("idx_opp_atoms_snap", "opp_demand_atoms", "snap_id"),
|
||||
("idx_opp_std_cluster", "opp_std_demands", "cluster_id, tier"),
|
||||
("idx_opp_domains_cluster", "opp_demand_domains", "cluster_id")]:
|
||||
try:
|
||||
await sor.sqlExe("CREATE INDEX IF NOT EXISTS %s ON %s (%s)" % (name, tbl, cols), {})
|
||||
await sor.sqlExe("COMMIT", {})
|
||||
print("index OK:", name)
|
||||
except Exception as e:
|
||||
print("index skip:", name, str(e)[:60])
|
||||
|
||||
asyncio.run(main())
|
||||
Loading…
x
Reference in New Issue
Block a user