feat(mining): P1聚类算法验证脚本(真实数据)——爬虫拉全量需求→rag_client Bearer embedding(10条/批,3次退避)→VDB kNN(pagerows=30)→τ阈值并查集聚类→小组质心并入→簇质量报告;坑:cookie会话401(Secure标志+aiohttp拒发http),生产路径Bearer不受限;独立脚本须initEnv()注入password_decode
This commit is contained in:
parent
b10ef50661
commit
4ad5dfda81
267
scripts/p1_cluster_validate.py
Normal file
267
scripts/p1_cluster_validate.py
Normal file
@ -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))
|
||||
Loading…
x
Reference in New Issue
Block a user