From 1e64ab7e752065edb160b84fd1323ec343a5bd08 Mon Sep 17 00:00:00 2001 From: yumoqing Date: Sat, 12 Sep 2026 11:41:51 +0800 Subject: [PATCH] =?UTF-8?q?feat(normalize):=20P2-=E2=91=A2=E7=B1=BB?= =?UTF-8?q?=E5=86=85=E5=85=B1=E6=80=A7=E6=8F=90=E5=8F=96=E5=BC=95=E6=93=8E?= =?UTF-8?q?=E2=80=94=E2=80=94opp=5Fnormalize.py(=E5=8E=9F=E5=AD=90?= =?UTF-8?q?=E5=8C=96LLM=E5=90=88=E6=89=B9=E5=B9=B6=E5=8F=91=E2=86=92embedd?= =?UTF-8?q?ing+=CF=842=E5=B9=B6=E6=9F=A5=E9=9B=86=E5=80=99=E9=80=89?= =?UTF-8?q?=E7=BB=84=E2=86=92LLM=E8=A3=81=E5=86=B3=E5=BD=92=E4=B8=80(?= =?UTF-8?q?=E7=9B=B8=E4=BC=BC=E5=BA=A6=E5=8F=AA=E5=BD=93=E8=A7=A6=E5=8F=91?= =?UTF-8?q?=E5=99=A8)=E2=86=92=E8=83=BD=E5=8A=9B=E5=9F=9F=E5=88=86?= =?UTF-8?q?=E7=BB=84=E2=86=92=E8=A6=86=E7=9B=96=E7=8E=87=E4=B8=89=E5=B1=82?= =?UTF-8?q?core=E2=89=A50.6/ext=E2=89=A50.4/personal);3=E6=96=B0=E8=A1=A8a?= =?UTF-8?q?toms/std=5Fdemands/domains+clusters=E8=A1=A5normalize=5Fstatus/?= =?UTF-8?q?error=E5=88=97;2=E5=B7=A5=E5=85=B7normalize=5Fcluster/coverage?= =?UTF-8?q?=5Freport(org=E9=9A=94=E7=A6=BB+=E6=9C=AA=E5=BD=92=E4=B8=80?= =?UTF-8?q?=E5=85=88=E6=8A=A5=E9=94=99=E5=BC=95=E5=AF=BC);params=E7=A7=8D?= =?UTF-8?q?=E5=AD=905=E9=A1=B9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- init/data.json | 7 +- models/opp_clusters.json | 12 + models/opp_demand_atoms.json | 61 ++++ models/opp_demand_domains.json | 61 ++++ models/opp_std_demands.json | 81 +++++ mysql.ddl.sql | 101 ++++++ pipeline_opportunity/opp_ability.py | 41 +++ pipeline_opportunity/opp_normalize.py | 427 ++++++++++++++++++++++++++ scripts/p2_create_tables.py | 42 +++ 9 files changed, 832 insertions(+), 1 deletion(-) create mode 100644 models/opp_demand_atoms.json create mode 100644 models/opp_demand_domains.json create mode 100644 models/opp_std_demands.json create mode 100644 pipeline_opportunity/opp_normalize.py create mode 100644 scripts/p2_create_tables.py diff --git a/init/data.json b/init/data.json index c3b173e..b193b8e 100644 --- a/init/data.json +++ b/init/data.json @@ -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": "扩展共性覆盖率阈值"} ] } diff --git a/models/opp_clusters.json b/models/opp_clusters.json index deba5dc..31eb7d5 100644 --- a/models/opp_clusters.json +++ b/models/opp_clusters.json @@ -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": "创建时间", diff --git a/models/opp_demand_atoms.json b/models/opp_demand_atoms.json new file mode 100644 index 0000000..95d57a2 --- /dev/null +++ b/models/opp_demand_atoms.json @@ -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" + } + ] +} diff --git a/models/opp_demand_domains.json b/models/opp_demand_domains.json new file mode 100644 index 0000000..8152546 --- /dev/null +++ b/models/opp_demand_domains.json @@ -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" + } + ] +} diff --git a/models/opp_std_demands.json b/models/opp_std_demands.json new file mode 100644 index 0000000..1f33ed6 --- /dev/null +++ b/models/opp_std_demands.json @@ -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" + } + ] +} diff --git a/mysql.ddl.sql b/mysql.ddl.sql index 8c12a72..bb04588 100644 --- a/mysql.ddl.sql +++ b/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 diff --git a/pipeline_opportunity/opp_ability.py b/pipeline_opportunity/opp_ability.py index 26d7e0c..a225213 100644 --- a/pipeline_opportunity/opp_ability.py +++ b/pipeline_opportunity/opp_ability.py @@ -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, } diff --git a/pipeline_opportunity/opp_normalize.py b/pipeline_opportunity/opp_normalize.py new file mode 100644 index 0000000..963a096 --- /dev/null +++ b/pipeline_opportunity/opp_normalize.py @@ -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, + } diff --git a/scripts/p2_create_tables.py b/scripts/p2_create_tables.py new file mode 100644 index 0000000..069994c --- /dev/null +++ b/scripts/p2_create_tables.py @@ -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())