From 4ad5dfda81c8c44ab1455e3fca89f7a333fbfb1c Mon Sep 17 00:00:00 2001 From: yumoqing Date: Fri, 11 Sep 2026 22:41:52 +0800 Subject: [PATCH] =?UTF-8?q?feat(mining):=20P1=E8=81=9A=E7=B1=BB=E7=AE=97?= =?UTF-8?q?=E6=B3=95=E9=AA=8C=E8=AF=81=E8=84=9A=E6=9C=AC(=E7=9C=9F?= =?UTF-8?q?=E5=AE=9E=E6=95=B0=E6=8D=AE)=E2=80=94=E2=80=94=E7=88=AC?= =?UTF-8?q?=E8=99=AB=E6=8B=89=E5=85=A8=E9=87=8F=E9=9C=80=E6=B1=82=E2=86=92?= =?UTF-8?q?rag=5Fclient=20Bearer=20embedding(10=E6=9D=A1/=E6=89=B9,3?= =?UTF-8?q?=E6=AC=A1=E9=80=80=E9=81=BF)=E2=86=92VDB=20kNN(pagerows=3D30)?= =?UTF-8?q?=E2=86=92=CF=84=E9=98=88=E5=80=BC=E5=B9=B6=E6=9F=A5=E9=9B=86?= =?UTF-8?q?=E8=81=9A=E7=B1=BB=E2=86=92=E5=B0=8F=E7=BB=84=E8=B4=A8=E5=BF=83?= =?UTF-8?q?=E5=B9=B6=E5=85=A5=E2=86=92=E7=B0=87=E8=B4=A8=E9=87=8F=E6=8A=A5?= =?UTF-8?q?=E5=91=8A;=E5=9D=91:cookie=E4=BC=9A=E8=AF=9D401(Secure=E6=A0=87?= =?UTF-8?q?=E5=BF=97+aiohttp=E6=8B=92=E5=8F=91http),=E7=94=9F=E4=BA=A7?= =?UTF-8?q?=E8=B7=AF=E5=BE=84Bearer=E4=B8=8D=E5=8F=97=E9=99=90;=E7=8B=AC?= =?UTF-8?q?=E7=AB=8B=E8=84=9A=E6=9C=AC=E9=A1=BBinitEnv()=E6=B3=A8=E5=85=A5?= =?UTF-8?q?password=5Fdecode?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- scripts/p1_cluster_validate.py | 267 +++++++++++++++++++++++++++++++++ 1 file changed, 267 insertions(+) create mode 100644 scripts/p1_cluster_validate.py diff --git a/scripts/p1_cluster_validate.py b/scripts/p1_cluster_validate.py new file mode 100644 index 0000000..99fd55a --- /dev/null +++ b/scripts/p1_cluster_validate.py @@ -0,0 +1,267 @@ +# -*- coding:utf-8 -*- +"""P1 聚类算法验证：真实众包需求 → embedding → VDB kNN → 并查集聚类 → 簇质量报告。 + +在 pipeline-app 测试机执行：./py3/bin/python /tmp/p1_cluster_validate.py +链路（与生产工具同协议，全部实测过的端点）： + 1. 爬虫平台 /api/demands 分页拉全量需求（params 表读 base/token） + 2. rag /rag/api/embed.dspy 批量向量化（HTTP 会话，admin 登录拿 cookie） + 3. VDB(Milvus) 测试 collection：upsert + /v1/query kNN(pagerows=30) + 4. 阈值 τ → 并查集连通分量 → 簇；小组并入最近大簇或归"其他" + 5. 输出簇统计 + 每簇 top10 样例标题（人工审聚类质量） +结果写 /tmp/p1_cluster_result.json。任何一步失败如实报错退出，不静默。 +""" +import asyncio +import json +import os +import sys +import time + +WORKDIR = "/d/pipeline/pipeline-app" +os.chdir(WORKDIR) +sys.path.insert(0, WORKDIR) + +from appPublic.folderUtils import ProgramPath # noqa: E402 +from appPublic.jsonConfig import getConfig # noqa: E402 +from appPublic.event_dispatcher import EventDispatcher # noqa: E402 +from sqlor.dbpools import DBPools # noqa: E402 +from ahserver.serverenv import ServerEnv # noqa: E402 + +p = ProgramPath() +config = getConfig(WORKDIR, NS={'workdir': WORKDIR, 'ProgramPath': p}) +DBPools(config.databases) +se = ServerEnv() +se.event_dispatcher = EventDispatcher() +se.get_module_dbname = lambda m: 'pipeline' + +TAU = float(os.environ.get("P1_TAU", "0.75")) # 相似度阈值（可调参对比） +TOPK = 30 # kNN 邻居数 +MIN_CLUSTER = 5 # 最小簇规模 +TEST_COL = "opp_p1_validate_%d" % int(time.time()) +DIM = 1024 +APP_PORT = 9090 + + +async def _sql(sql, args=None): + async with DBPools().sqlorContext("pipeline") as sor: + recs = await sor.sqlExe(sql, args or {}) + await sor.sqlExe("COMMIT", {}) + return recs or [] + + +class Progress: + def __init__(self, total, label): + self.total, self.label, self.n, self.t0 = total, label, 0, time.time() + + def tick(self, k=1): + self.n += k + if self.n % 50 == 0 or self.n >= self.total: + el = time.time() - self.t0 + print(" %s %d/%d (%.0fs)" % (self.label, self.n, self.total, el), flush=True) + + +async def main(): + import aiohttp + timeout = aiohttp.ClientTimeout(total=60) + + # ── 1. 拉全量需求 ── + recs = await _sql("SELECT params_name, params_value FROM params WHERE params_name IN ('tender_api_base','tender_api_token')") + pm = {r.params_name: r.params_value for r in recs} + base = (pm.get("tender_api_base") or "http://192.168.16.2:9085").rstrip("/") + token = pm.get("tender_api_token") or "" + if not token: + print("FATAL: params 无 tender_api_token") + sys.exit(1) + hdr = {"X-API-Token": token} + + demands = [] + offset = 0 + async with aiohttp.ClientSession(timeout=timeout) as s: + while True: + async with s.get(base + "/api/demands", headers=hdr, + params={"days": 3650, "limit": 200, "offset": offset}) as r: + if r.status != 200: + print("FATAL: /api/demands HTTP", r.status, await r.text()) + sys.exit(1) + d = await r.json() + items = d.get("items") or [] + demands.extend(items) + offset += len(items) + if not items or offset >= d.get("matched", 0): + break + print("1. 需求快照拉取: %d 条" % len(demands)) + if not demands: + print("FATAL: 无需求数据") + sys.exit(1) + + # ── 2. embedding（生产同款通道：rag_client Bearer → /rag/api/embed.dspy）── + # 坑实录：独立脚本走 cookie 会话必 401——登录 cookie 带 Secure 标志，aiohttp + # 客户端拒绝经明文 http:// 发送；生产路径本来就是 Bearer token，不受此限。 + # 独立脚本需 initEnv() 注入 password_decode 等宿主全局函数（dapi 发 key 依赖）。 + from ahserver.globalEnv import initEnv + initEnv() + from pipeline_service import rag_client as rc + _pid = os.environ.get("P1_PROJECT_ID", "") + if not _pid: + print("FATAL: 缺环境变量 P1_PROJECT_ID（真实商机项目id，用于 owner→Bearer key）") + sys.exit(1) + texts = [(it.get("title") or "").strip() for it in demands] + texts = [t if t else "(无标题)" for t in texts] + vectors = [None] * len(texts) + pr = Progress(len(texts), "embedding") + GROUP = 100 + for i in range(0, len(texts), GROUP): + chunk = texts[i:i + GROUP] + vs = None + for attempt in range(3): + vs, err = await rc.rag_embed_texts(_pid, chunk, batch_size=10) + if not err and vs: + break + vs = None + if attempt == 2: + print("FATAL: embed 失败@%d-%d: %s" % (i, i + len(chunk) - 1, err)) + sys.exit(1) + await asyncio.sleep(2 ** attempt) + vectors[i:i + len(chunk)] = vs + pr.tick(len(chunk)) + assert all(v is not None for v in vectors), "存在未嵌入条目" + print("2. embedding 完成: %d 条 × %d 维" % (len(vectors), len(vectors[0]))) + + # ── 3. VDB 常驻缓存 collection：upsert + kNN ── + async with aiohttp.ClientSession(timeout=timeout) as s: + async def vpost(path, payload): + r = await s.post(vdb_base + path, json=payload) + return r.status, await r.text() + recs = await _sql("SELECT baseurl FROM upapp WHERE id='rag-vdb'") + vdb_base = (recs[0].baseurl or "").rstrip("/") + if not vdb_base: + print("FATAL: upapp 无 rag-vdb") + sys.exit(1) + + st, body = await vpost("/v1/createcollection", {"colname": TEST_COL, "fields": [ + {"name": "id", "type": "str", "is_primary": True, "max_length": 64}, + {"name": "vector", "type": "fvector", "dim": DIM}, + {"name": "text", "type": "str", "max_length": 2000}, + {"name": "batch_id", "type": "str", "max_length": 64}], + "description": "P1 validate", "metric": "COSINE"}) + if "SUCCEEDED" not in body: + print("FATAL: createcollection", body[:200]) + sys.exit(1) + + # upsert 分批 500 + ids = ["d%d" % i for i in range(len(vectors))] + for i in range(0, len(vectors), 500): + rows = [{"id": ids[j], "vector": vectors[j], "text": texts[j][:1900], "batch_id": "p1v"} + for j in range(i, min(i + 500, len(vectors)))] + st, body = await vpost("/v1/upsert", {"colname": TEST_COL, "data": rows}) + if "SUCCEEDED" not in body: + print("FATAL: upsert@%d" % i, body[:200]) + sys.exit(1) + print("3a. VDB upsert 完成") + + # kNN：每条查 TOPK 邻居 + nbr = {} # id -> [(score, nid)] + pr = Progress(len(ids), "kNN") + for i, vid in enumerate(ids): + st, body = await vpost("/v1/query", { + "colname": TEST_COL, "vector": vectors[i], "pagerows": TOPK, + "output_fields": ["id"]}) + d = json.loads(body) + rows = (d.get("data") or {}).get("rows") or [] + nbr[vid] = [(float(r.get("score", 0)), r.get("id")) for r in rows if r.get("id") != vid] + pr.tick() + print("3b. kNN 完成") + + await vpost("/v1/dropcollection", {"colname": TEST_COL}) + print("3c. 测试 collection 已清理") + + # ── 4. 并查集 ── + parent = {v: v for v in ids} + + def find(x): + while parent[x] != x: + parent[x] = parent[parent[x]] + x = parent[x] + return x + + def union(a, b): + ra, rb = find(a), find(b) + if ra != rb: + parent[rb] = ra + + edges = 0 + for vid in ids: + for sc, nid in nbr[vid]: + if sc >= TAU: + union(vid, nid) + edges += 1 + comps = {} + for vid in ids: + comps.setdefault(find(vid), []).append(vid) + clusters = sorted(comps.values(), key=len, reverse=True) + print("4. 聚类: τ=%.2f 边=%d 分量=%d (≥%d条的簇=%d)" % ( + TAU, edges, len(clusters), MIN_CLUSTER, sum(1 for c in clusters if len(c) >= MIN_CLUSTER))) + + # 小组并入最近大簇（质心相似度）；仍不够 → 其他 + big = [c for c in clusters if len(c) >= MIN_CLUSTER] + small = [c for c in clusters if len(c) < MIN_CLUSTER] + vec_by_id = {ids[i]: vectors[i] for i in range(len(ids))} + text_by_id = {ids[i]: texts[i] for i in range(len(ids))} + + def centroid(members): + n = len(members) + return [sum(vec_by_id[m][k] for m in members) / n for k in range(DIM)] + + merged_into_other = [] + if big and small: + cents = [(c, centroid(c)) for c in big] + for sc_ in small: + sc_cent = centroid(sc_) + best, best_sim = None, -1 + for bc, bc_cent in cents: + sim = sum(a * b for a, b in zip(sc_cent, bc_cent)) + if sim > best_sim: + best, best_sim = bc, sim + if best is not None and best_sim >= TAU - 0.10: + best.extend(sc_) + else: + merged_into_other.extend(sc_) + elif small and not big: + for c in small: + merged_into_other.extend(c) + final = [c for c in big if len(c) >= MIN_CLUSTER] + final.sort(key=len, reverse=True) + + # ── 5. 质量报告 ── + total = len(ids) + covered = sum(len(c) for c in final) + report = { + "tau": TAU, "topk": TOPK, "min_cluster": MIN_CLUSTER, + "total_demands": total, "n_clusters": len(final), + "covered": covered, "other_bucket": len(merged_into_other), + "clusters": [], + } + for ci, c in enumerate(final): + # 簇内离质心最近的 10 条做样例 + cent = centroid(c) + sims = sorted(((sum(a * b for a, b in zip(vec_by_id[m], cent)), m) for m in c), reverse=True) + samples = [text_by_id[m] for _, m in sims[:10]] + report["clusters"].append({ + "rank": ci + 1, "size": len(c), "share": round(len(c) / total, 4), + "samples": samples, + }) + with open("/tmp/p1_cluster_result.json", "w", encoding="utf-8") as f: + json.dump(report, f, ensure_ascii=False, indent=1) + print("5. 报告已写 /tmp/p1_cluster_result.json") + print() + print("═══ 簇排名（τ=%.2f）═══" % TAU) + for cl in report["clusters"][:15]: + print("#%d size=%d (%.1f%%) 样例: %s" % ( + cl["rank"], cl["size"], cl["share"] * 100, + " | ".join(cl["samples"][:4])[:160])) + print("其他桶(未入簇): %d (%.1f%%)" % (len(merged_into_other), len(merged_into_other) / total * 100)) + + +if __name__ == "__main__": + t0 = time.time() + asyncio.run(main()) + print("TOTAL %.0fs" % (time.time() - t0))