268 lines
11 KiB
Python
268 lines
11 KiB
Python
# -*- 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))
|