857 lines
39 KiB
Python
857 lines
39 KiB
Python
"""Super Agent — 知识图谱馆长（全城网格自治采集，常驻不停）。
|
||
|
||
科学合理 & 省额度的核心：**地理网格扫描**，不再用"热度关键词"。
|
||
• 把贵阳整个城市切成网格，逐格用高德"多边形(矩形)搜索 + 官方类型编码"
|
||
系统性扫，每格每页都是不同地理切片 → 几乎不重复 → 不浪费 API 额度。
|
||
• 每格扫到第几页**持久化**（gaode_grid_cells.next_page），跨步/跨轮接着扫，
|
||
扫尽的格标记 exhausted **永不重复请求**（省额度的关键）。
|
||
• 稠密区一格出不完 → **自适应四叉细分**，保证不漏。
|
||
• 全城扫完不"结束"，转**驻守巡检**（不再消耗额度，仅响应停止/再播种）。
|
||
• 进度=网格地理覆盖率，真实可反映，非拍脑袋数字。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import re
|
||
|
||
from app.api.graph import _get_graph
|
||
from app.config import settings
|
||
from app.agents.gaode_connector import AMAP_TYPECODES, search_polygon
|
||
from app.agents.super_ingest import ingest_rows
|
||
from app.agents.distill_gate import distill_entity, ATTR_FIELDS
|
||
from app.agents.web_agent import web_enrich
|
||
from app.agents.xhs_agent import xhs_enrich
|
||
from app.agents.douyin_agent import douyin_enrich
|
||
from app.agents.event_miner import mine_events
|
||
from app.db import (
|
||
sa_append_step, sa_finish, sa_stop_requested, sa_add_task, sa_set_status,
|
||
sa_has_open_escalation, create_acquisition_task, create_notification,
|
||
get_admin_user_id, grid_seed, grid_counts, grid_pending_cats,
|
||
grid_take_next, grid_update, grid_subdivide,
|
||
sa_merge_candidate_payload, sa_record_conflict, sa_record_schema_proposal,
|
||
)
|
||
|
||
_BG: set = set()
|
||
|
||
# 贵阳整市外接框 (min_lng, min_lat, max_lng, max_lat)：含主城+周边区县
|
||
_GY_BBOX = (106.20, 26.10, 107.30, 27.05)
|
||
_CELL = 0.08 # 初始网格边长(°)，约 8km
|
||
_OFFSET = 25 # 高德多边形单页上限
|
||
_MAX_PAGE_SPLIT = 8 # 单格翻到第N页仍满 → 太密 → 四叉细分
|
||
_MAX_DEPTH = 3 # 最深细分层级（0.08→约1km格）
|
||
_API_PACING = 0.35 # 每次高德调用间隔(秒)，护 QPS
|
||
_HARD_MAX_STEPS = 100000 # 仅极端兜底（持久化，跨轮续扫，不暴露用户）
|
||
_MAX_API_PER_RUN = 6000 # 单进程跑安全上限；命中转驻守，下轮自动续
|
||
_STEWARD_INTERVAL = 1800
|
||
_STEWARD_TICK = 10
|
||
_ENRICH_BATCH = 5 # 每步最多蒸馏富集多少个实体
|
||
_ENRICH_EVERY = 4 # 网格忙时每 N 步插一次知识富集
|
||
_WEB_BATCH = 2 # 每步最多联网富集多少个实体(真浏览器较慢)
|
||
_XHS_BATCH = 1 # 小红书浏览器很慢/脆，每步只采 1 个
|
||
|
||
|
||
def _place_counts() -> dict:
|
||
g = _get_graph()
|
||
out: dict = {}
|
||
try:
|
||
for row in g.query(
|
||
"MATCH (p:Place) RETURN coalesce(p.place_type,'?'), count(*)"
|
||
).result_set:
|
||
out[row[0]] = row[1]
|
||
except Exception:
|
||
pass
|
||
return out
|
||
|
||
|
||
def _coverage() -> dict:
|
||
from app.agents.super_ingest import _PT
|
||
counts = _place_counts()
|
||
cat2pt = {cat: _PT.get(cat, "poi") for cat in AMAP_TYPECODES}
|
||
items = [{"cat": cat, "place_type": cat2pt[cat],
|
||
"current": int(counts.get(cat2pt[cat], 0))}
|
||
for cat in AMAP_TYPECODES]
|
||
return {"items": items, "total": sum(i["current"] for i in items)}
|
||
|
||
|
||
def _grid_for(bbox: tuple, step: float) -> list[tuple]:
|
||
mnlng, mnlat, mxlng, mxlat = bbox
|
||
cells, lng = [], mnlng
|
||
while lng < mxlng:
|
||
lat = mnlat
|
||
nlng = round(min(lng + step, mxlng), 6)
|
||
while lat < mxlat:
|
||
nlat = round(min(lat + step, mxlat), 6)
|
||
cells.append((round(lng, 6), round(lat, 6), nlng, nlat))
|
||
lat = nlat
|
||
lng = nlng
|
||
return cells
|
||
|
||
|
||
def _quads(c: dict) -> list[tuple]:
|
||
"""父格四等分为 4 个子格(矩形对半切)。"""
|
||
mnlng, mnlat = c["min_lng"], c["min_lat"]
|
||
mxlng, mxlat = c["max_lng"], c["max_lat"]
|
||
midlng = round((mnlng + mxlng) / 2, 6)
|
||
midlat = round((mnlat + mxlat) / 2, 6)
|
||
return [
|
||
(mnlng, mnlat, midlng, midlat), (midlng, mnlat, mxlng, midlat),
|
||
(mnlng, midlat, midlng, mxlat), (midlng, midlat, mxlng, mxlat),
|
||
]
|
||
|
||
|
||
async def _seed_if_needed() -> None:
|
||
have = await grid_counts()
|
||
for cat, tc in AMAP_TYPECODES.items():
|
||
if cat not in have:
|
||
await grid_seed(cat, tc, _grid_for(_GY_BBOX, _CELL))
|
||
|
||
|
||
async def _escalate(run_id: int, step: int, cat: str, cur: int) -> None:
|
||
if await sa_has_open_escalation(cat):
|
||
await sa_add_task(run_id, step, cat, "escalate",
|
||
"全城网格已扫尽，该类仍偏少", "skip_dup",
|
||
status="escalated", note="已有未结工单，不重复打扰")
|
||
return
|
||
task_id = None
|
||
try:
|
||
task = await create_acquisition_task({
|
||
"tenant_id": settings.default_tenant,
|
||
"project_id": settings.default_project,
|
||
"created_by": "super_agent",
|
||
"title": f"【Super Agent 求助】「{cat}」全城网格已扫尽仍偏少",
|
||
"description": (
|
||
f"馆长已对贵阳全城做网格化高德采集，「{cat}」当前仅 {cur} 条，"
|
||
f"高德官方源对该类覆盖有限。建议人工核查类型编码或改用"
|
||
f"小红书/大众点评/官方名录等渠道补全。馆长继续驻守，"
|
||
f"工单结清/数据增长后自动恢复。"),
|
||
"scenario_tags": ["super_agent", "escalation", cat],
|
||
"target_entity_types": ["Place"],
|
||
"target_fields": [],
|
||
"suggested_collection_method": "manual_or_alt_source",
|
||
"priority": 1,
|
||
})
|
||
task_id = task["id"]
|
||
except Exception:
|
||
task_id = None
|
||
try:
|
||
uid = await get_admin_user_id()
|
||
if uid:
|
||
await create_notification(
|
||
uid, title=f"Super Agent 求助：「{cat}」全城网格已扫尽",
|
||
body=(f"「{cat}」仅 {cur} 条，高德源覆盖有限。已开工单"
|
||
+ (f" #{task_id}" if task_id else "")
|
||
+ "，请人工/改渠道；馆长继续驻守。"),
|
||
ntype="task", related_task_id=task_id)
|
||
except Exception:
|
||
pass
|
||
await sa_add_task(run_id, step, cat, "escalate",
|
||
"全城网格已扫尽，该类仍偏少", "notify_admin",
|
||
status="escalated", related_task_id=task_id)
|
||
await sa_append_step(run_id, {
|
||
"step": step, "action": "escalate", "cat": cat,
|
||
"reason": f"「{cat}」全城网格扫尽仅{cur}条，已开工单并通知管理员，继续驻守"})
|
||
|
||
|
||
def _enrich_targets(limit: int) -> list[dict]:
|
||
"""未富集过的 Place + 锚点(名/址/区/类) + 现有软字段。
|
||
|
||
取数显式排除 经纬度/电话（隐私红线，绝不外发给蒸馏模型）。
|
||
"""
|
||
g = _get_graph()
|
||
try:
|
||
rs = g.query(
|
||
"MATCH (p:Place) WHERE p.enrich_done IS NULL "
|
||
"RETURN p.element_id, p.name, coalesce(p.place_type,''), "
|
||
"coalesce(p.district,''), coalesce(p.address,''), "
|
||
"coalesce(p.summary,''), coalesce(p.history,''), "
|
||
"coalesce(p.features,''), coalesce(p.suitable_for,''), "
|
||
"coalesce(p.best_season,''), coalesce(p.ticket_hint,'') "
|
||
"LIMIT $n", {"n": limit}).result_set
|
||
except Exception:
|
||
return []
|
||
out = []
|
||
for r in rs:
|
||
if not (r and r[1]):
|
||
continue
|
||
existing = {k: v for k, v in zip(
|
||
ATTR_FIELDS, [r[5], r[6], r[7], r[8], r[9], r[10]]) if v}
|
||
out.append({"eid": r[0], "name": r[1], "place_type": r[2],
|
||
"district": r[3], "address": r[4],
|
||
"existing": existing})
|
||
return out
|
||
|
||
|
||
def _apply_enrich(eid: str, fields: dict) -> None:
|
||
"""共识字段写回 FalkorDB 节点；无论是否有字段都打 enrich_done 防重复空跑。"""
|
||
g = _get_graph()
|
||
sets = ["p.enrich_done=1"]
|
||
params = {"eid": eid}
|
||
for k in ATTR_FIELDS:
|
||
if fields.get(k):
|
||
sets.append(f"p.{k}=${k}")
|
||
params[k] = fields[k]
|
||
try:
|
||
g.query(f"MATCH (p:Place {{element_id:$eid}}) SET {','.join(sets)}",
|
||
params)
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
async def _distill_enrich(run_id: int, step: int,
|
||
targets: list[dict]) -> bool:
|
||
"""工具：多模型知识蒸馏，给已有实体补"知识层"属性。
|
||
|
||
独立数据来源（不是高德质检闸门）：问 N 个模型脑内知识 → 全局模型聚合
|
||
跨模型共识 → 写回 FalkorDB 节点 + 候选 payload（审计可溯）。
|
||
"""
|
||
enriched = adopt_fields = keep_total = conflict_total = uncertain_total = 0
|
||
conflict_names: list[str] = []
|
||
last = ""
|
||
for t in targets:
|
||
res = await distill_entity(t)
|
||
last = res.get("summary", "")
|
||
if not res.get("ok"):
|
||
# 配置/连通问题：整批中止——不打标记(待配置后重试)、不空转
|
||
await sa_add_task(
|
||
run_id, step, "蒸馏", "distill",
|
||
f"知识蒸馏未就绪：{last}", "distill_enrich",
|
||
result={"fetched": 0, "approved": 0, "pending": 0,
|
||
"skipped": 0}, status="skipped", note=last)
|
||
await sa_append_step(run_id, {
|
||
"step": step, "action": "distill",
|
||
"reason": f"知识蒸馏未配置/不可用（{last}），本步跳过，"
|
||
f"配好≥2蒸馏模型+全局聚合后自动恢复"})
|
||
return False
|
||
|
||
adopt = res.get("adopt") or {}
|
||
_apply_enrich(t["eid"], adopt) # 只写 adopt；无则仅打标记防空转
|
||
if adopt:
|
||
await sa_merge_candidate_payload(t["eid"], adopt)
|
||
enriched += 1
|
||
adopt_fields += len(adopt)
|
||
keep_total += len(res.get("keep") or [])
|
||
uncertain_total += len(res.get("uncertain") or [])
|
||
|
||
# 与图谱既有值矛盾 → 不覆盖，落 validation_issues 转人工
|
||
for c in (res.get("conflict") or []):
|
||
await sa_record_conflict(
|
||
t["eid"], c.get("field", ""), c.get("existing", ""),
|
||
c.get("distilled", ""), c.get("note", ""))
|
||
conflict_total += 1
|
||
if t.get("name"):
|
||
conflict_names.append(f"{t['name']}·{c.get('field','')}")
|
||
|
||
res_obj = {"fetched": len(targets), "approved": adopt_fields,
|
||
"pending": conflict_total, "skipped": uncertain_total}
|
||
await sa_add_task(
|
||
run_id, step, "蒸馏", "distill",
|
||
f"多模型知识蒸馏富集 {len(targets)} 个实体（{last}）",
|
||
"distill_enrich", result=res_obj, status="done",
|
||
note=f"富集{enriched}实体/{adopt_fields}字段·一致{keep_total}"
|
||
f"·矛盾{conflict_total}·存疑{uncertain_total}")
|
||
await sa_append_step(run_id, {
|
||
"step": step, "action": "distill",
|
||
"reason": f"知识蒸馏富集：{enriched}/{len(targets)} 实体补 {adopt_fields} 字段，"
|
||
f"矛盾 {conflict_total} 转人工（{last}）"})
|
||
|
||
if conflict_total:
|
||
try:
|
||
uid = await get_admin_user_id()
|
||
if uid:
|
||
await create_notification(
|
||
uid,
|
||
title=f"蒸馏发现 {conflict_total} 处与图谱既有数据矛盾",
|
||
body=("需人工裁决（蒸馏未覆盖图谱）："
|
||
+ "、".join(conflict_names[:8])
|
||
+ ("…" if len(conflict_names) > 8 else "")
|
||
+ "。详见 数据质量 / 校验问题(distill_conflict)。"),
|
||
ntype="task")
|
||
except Exception:
|
||
pass
|
||
return True
|
||
|
||
|
||
def _web_targets(limit: int) -> list[dict]:
|
||
"""未联网富集过的 Place（无 web_done）+ 锚点 + 现有软字段。
|
||
取数显式排除 经纬度/电话（隐私红线）。"""
|
||
g = _get_graph()
|
||
try:
|
||
# 景点最可能有公开百科页 → 优先联网富集，命中率最高
|
||
rs = g.query(
|
||
"MATCH (p:Place) WHERE p.web_done IS NULL "
|
||
"RETURN p.element_id, p.name, coalesce(p.place_type,''), "
|
||
"coalesce(p.district,''), coalesce(p.address,''), "
|
||
"coalesce(p.summary,''), coalesce(p.history,''), "
|
||
"coalesce(p.features,''), coalesce(p.suitable_for,''), "
|
||
"coalesce(p.best_season,''), coalesce(p.ticket_hint,'') "
|
||
"ORDER BY CASE WHEN p.place_type='sight' THEN 0 ELSE 1 END, "
|
||
"p.element_id LIMIT $n", {"n": limit}).result_set
|
||
except Exception:
|
||
return []
|
||
out = []
|
||
for r in rs:
|
||
if not (r and r[1]):
|
||
continue
|
||
existing = {k: v for k, v in zip(
|
||
ATTR_FIELDS, [r[5], r[6], r[7], r[8], r[9], r[10]]) if v}
|
||
out.append({"eid": r[0], "name": r[1], "place_type": r[2],
|
||
"district": r[3], "address": r[4], "existing": existing})
|
||
return out
|
||
|
||
|
||
def _apply_web(eid: str, adopt: dict) -> None:
|
||
"""网页采纳字段写回；打 web_done(必)；有 adopt 则连 enrich_done 一并打
|
||
(网页权威，省一次记忆蒸馏)。"""
|
||
g = _get_graph()
|
||
sets = ["p.web_done=1"]
|
||
params = {"eid": eid}
|
||
if adopt:
|
||
sets.append("p.enrich_done=1")
|
||
for k in ATTR_FIELDS:
|
||
if adopt.get(k):
|
||
sets.append(f"p.{k}=${k}")
|
||
params[k] = adopt[k]
|
||
try:
|
||
g.query(f"MATCH (p:Place {{element_id:$eid}}) SET {','.join(sets)}",
|
||
params)
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
async def _web_enrich(run_id: int, step: int, targets: list[dict]) -> bool:
|
||
"""工具：browser-use 式联网采集（真浏览器抓权威页 → opus 抽取对齐）。"""
|
||
enriched = adopt_fields = found = conflict_total = gap_total = 0
|
||
last = ""
|
||
for t in targets:
|
||
r = await web_enrich(t)
|
||
last = r.get("summary", "")
|
||
if not r.get("ok"):
|
||
await sa_add_task(run_id, step, "联网", "web",
|
||
f"web_agent 未就绪：{last}", "web_agent",
|
||
result={"fetched": 0, "approved": 0,
|
||
"pending": 0, "skipped": 0},
|
||
status="skipped", note=last)
|
||
await sa_append_step(run_id, {
|
||
"step": step, "action": "web",
|
||
"reason": f"web_agent 未配置/不可用（{last}），跳过待恢复"})
|
||
return False
|
||
if not r.get("found") or r.get("entity_match") is False:
|
||
_apply_web(t["eid"], {}) # 标记 web_done，不再重复抓
|
||
continue
|
||
found += 1
|
||
adopt = r.get("adopt") or {}
|
||
_apply_web(t["eid"], adopt)
|
||
if adopt:
|
||
await sa_merge_candidate_payload(t["eid"], adopt)
|
||
enriched += 1
|
||
adopt_fields += len(adopt)
|
||
for c in (r.get("conflict") or []):
|
||
await sa_record_conflict(t["eid"], c.get("field", ""),
|
||
c.get("existing", ""),
|
||
c.get("web", ""), c.get("note", ""))
|
||
conflict_total += 1
|
||
for sgap in (r.get("schema_gaps") or []):
|
||
iid = await sa_record_schema_proposal(
|
||
sgap.get("attr", ""), sgap.get("field", "")
|
||
or re.sub(r"\W+", "_", sgap.get("attr", "")).strip("_"),
|
||
str(sgap.get("value", ""))[:200],
|
||
f"web_agent 在「{t['name']}」网页发现：{sgap.get('why','')}",
|
||
float(r.get("confidence") or 0.7))
|
||
if iid:
|
||
gap_total += 1
|
||
|
||
res_obj = {"fetched": len(targets), "approved": adopt_fields,
|
||
"pending": conflict_total, "skipped": gap_total}
|
||
await sa_add_task(
|
||
run_id, step, "联网", "web",
|
||
f"browser-use 联网采集 {len(targets)} 个实体（{last}）",
|
||
"web_agent", result=res_obj, status="done",
|
||
note=f"命中{found}·补{adopt_fields}字段·矛盾{conflict_total}"
|
||
f"·schema提案{gap_total}")
|
||
await sa_append_step(run_id, {
|
||
"step": step, "action": "web",
|
||
"reason": f"联网采集：命中 {found}/{len(targets)}，补 {adopt_fields} 字段，"
|
||
f"矛盾 {conflict_total} 转人工，schema 提案 {gap_total}（{last}）"})
|
||
if conflict_total or gap_total:
|
||
try:
|
||
uid = await get_admin_user_id()
|
||
if uid:
|
||
await create_notification(
|
||
uid,
|
||
title=f"web_agent：{conflict_total} 矛盾 / {gap_total} schema 提案待裁决",
|
||
body="联网采集与图谱矛盾或发现新属性，详见 数据质量/校验问题 "
|
||
"与 本体建模/字段提案。",
|
||
ntype="task")
|
||
except Exception:
|
||
pass
|
||
return True
|
||
|
||
|
||
def _xhs_targets(limit: int) -> list[dict]:
|
||
"""未采过小红书(无 xhs_done)的 Place；美食最优先(小红书食/玩为主)。"""
|
||
g = _get_graph()
|
||
try:
|
||
rs = g.query(
|
||
"MATCH (p:Place) WHERE p.xhs_done IS NULL "
|
||
"RETURN p.element_id, p.name, coalesce(p.place_type,''), "
|
||
"coalesce(p.district,'') "
|
||
"ORDER BY CASE WHEN p.place_type='eat' THEN 0 "
|
||
"WHEN p.place_type='sight' THEN 1 ELSE 2 END, p.element_id "
|
||
"LIMIT $n", {"n": limit}).result_set
|
||
except Exception:
|
||
return []
|
||
return [{"eid": r[0], "name": r[1], "place_type": r[2], "district": r[3]}
|
||
for r in rs if r and r[1]]
|
||
|
||
|
||
def _apply_xhs(eid: str, tags: list[str]) -> None:
|
||
"""体验标签写回：MERGE ExperienceTag + (Place)-[:HAS_TAG]->(tag)；
|
||
无论有无标签都打 xhs_done 防重复。"""
|
||
g = _get_graph()
|
||
try:
|
||
g.query("MATCH (p:Place {element_id:$eid}) SET p.xhs_done=1",
|
||
{"eid": eid})
|
||
for t in tags:
|
||
g.query(
|
||
"MATCH (p:Place {element_id:$eid}) "
|
||
"MERGE (e:ExperienceTag {name:$t}) "
|
||
"MERGE (p)-[:HAS_TAG]->(e) SET e.source='xiaohongshu'",
|
||
{"eid": eid, "t": t})
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
async def _xhs_enrich(run_id: int, step: int, targets: list[dict]) -> bool:
|
||
"""工具：小红书 UGC → 体验标签。未登录→升级人工(一次性登录)，本轮关闭。"""
|
||
done = tags_total = found = ev_total = 0
|
||
last = ""
|
||
for t in targets:
|
||
r = await xhs_enrich(t)
|
||
last = r.get("summary", "")
|
||
if not r.get("ok"):
|
||
await sa_add_task(run_id, step, "小红书", "xhs",
|
||
f"xhs_agent 未就绪：{last}", "xhs_agent",
|
||
result={"fetched": 0, "approved": 0,
|
||
"pending": 0, "skipped": 0},
|
||
status="skipped", note=last)
|
||
await sa_append_step(run_id, {"step": step, "action": "xhs",
|
||
"reason": f"小红书未配置/停用（{last}）"})
|
||
return False
|
||
if r.get("need_login"):
|
||
if not await sa_has_open_escalation("小红书登录"):
|
||
try:
|
||
task = await create_acquisition_task({
|
||
"tenant_id": settings.default_tenant,
|
||
"project_id": settings.default_project,
|
||
"created_by": "super_agent",
|
||
"title": "【Super Agent 求助】小红书需一次性人工登录",
|
||
"description": "后台 xhs_agent 检测到未登录。请在项目根目录运行 "
|
||
"`python3 scripts/xhs_login.py`，弹出浏览器里登录"
|
||
"小红书一次；cookie 持久化后馆长自动恢复采集。",
|
||
"scenario_tags": ["super_agent", "escalation", "小红书登录"],
|
||
"target_entity_types": ["Place"], "target_fields": [],
|
||
"suggested_collection_method": "manual_login",
|
||
"priority": 1})
|
||
uid = await get_admin_user_id()
|
||
if uid:
|
||
await create_notification(
|
||
uid, title="小红书需一次性登录",
|
||
body="运行 scripts/xhs_login.py 登录一次即可，馆长自动恢复。",
|
||
ntype="task", related_task_id=task["id"])
|
||
except Exception:
|
||
pass
|
||
await sa_add_task(run_id, step, "小红书", "xhs",
|
||
"小红书未登录，已升级人工一次性登录", "notify_admin",
|
||
result={"fetched": 0, "approved": 0,
|
||
"pending": 0, "skipped": 0},
|
||
status="escalated", note="待 scripts/xhs_login.py")
|
||
await sa_append_step(run_id, {
|
||
"step": step, "action": "xhs",
|
||
"reason": "小红书未登录 → 已开工单+通知(运行 xhs_login.py)，本轮暂停小红书"})
|
||
return False
|
||
ev_total += int(r.get("evidence_saved") or 0)
|
||
if r.get("found"):
|
||
_apply_xhs(t["eid"], r.get("tags") or [])
|
||
found += 1
|
||
if r.get("tags"):
|
||
done += 1
|
||
tags_total += len(r["tags"])
|
||
else:
|
||
_apply_xhs(t["eid"], []) # 标记，避免重复
|
||
res_obj = {"fetched": len(targets), "approved": tags_total,
|
||
"pending": ev_total, "skipped": len(targets) - found}
|
||
await sa_add_task(run_id, step, "小红书", "xhs",
|
||
f"小红书 UGC 采集 {len(targets)} 个实体（{last}）",
|
||
"xhs_agent", result=res_obj, status="done",
|
||
note=f"命中{found}·证据{ev_total}条·体验标签{tags_total}")
|
||
await sa_append_step(run_id, {
|
||
"step": step, "action": "xhs",
|
||
"reason": f"小红书采集：命中 {found}/{len(targets)}，证据入库 {ev_total} 条，"
|
||
f"产出 {tags_total} 个体验标签（{last}）"})
|
||
return True
|
||
|
||
|
||
def _dy_targets(limit: int) -> list[dict]:
|
||
"""未采过抖音(无 dy_done)的 Place；美食/景点优先(抖音食玩为主)。"""
|
||
g = _get_graph()
|
||
try:
|
||
rs = g.query(
|
||
"MATCH (p:Place) WHERE p.dy_done IS NULL "
|
||
"RETURN p.element_id, p.name, coalesce(p.place_type,''), "
|
||
"coalesce(p.district,'') "
|
||
"ORDER BY CASE WHEN p.place_type='eat' THEN 0 "
|
||
"WHEN p.place_type='sight' THEN 1 ELSE 2 END, p.element_id "
|
||
"LIMIT $n", {"n": limit}).result_set
|
||
except Exception:
|
||
return []
|
||
return [{"eid": r[0], "name": r[1], "place_type": r[2], "district": r[3]}
|
||
for r in rs if r and r[1]]
|
||
|
||
|
||
def _apply_dy(eid: str, tags: list[str]) -> None:
|
||
g = _get_graph()
|
||
try:
|
||
g.query("MATCH (p:Place {element_id:$eid}) SET p.dy_done=1",
|
||
{"eid": eid})
|
||
for t in tags:
|
||
g.query(
|
||
"MATCH (p:Place {element_id:$eid}) "
|
||
"MERGE (e:ExperienceTag {name:$t}) "
|
||
"MERGE (p)-[:HAS_TAG]->(e) SET e.source='douyin'",
|
||
{"eid": eid, "t": t})
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
async def _douyin_enrich(run_id: int, step: int,
|
||
targets: list[dict]) -> bool:
|
||
"""工具：抖音 UGC → 证据层+体验标签。未登录→升级一次性登录，本轮关闭。"""
|
||
done = tags_total = found = ev_total = 0
|
||
last = ""
|
||
for t in targets:
|
||
r = await douyin_enrich(t)
|
||
last = r.get("summary", "")
|
||
if not r.get("ok"):
|
||
await sa_add_task(run_id, step, "抖音", "douyin",
|
||
f"douyin_agent 未就绪：{last}", "douyin_agent",
|
||
result={"fetched": 0, "approved": 0,
|
||
"pending": 0, "skipped": 0},
|
||
status="skipped", note=last)
|
||
await sa_append_step(run_id, {"step": step, "action": "douyin",
|
||
"reason": f"抖音未配置/停用（{last}）"})
|
||
return False
|
||
if r.get("need_login"):
|
||
if not await sa_has_open_escalation("抖音登录"):
|
||
try:
|
||
task = await create_acquisition_task({
|
||
"tenant_id": settings.default_tenant,
|
||
"project_id": settings.default_project,
|
||
"created_by": "super_agent",
|
||
"title": "【Super Agent 求助】抖音需一次性人工登录",
|
||
"description": "后台 douyin_agent 检测到未登录。请在项目根目录"
|
||
"运行 `python3 scripts/douyin_login.py`，弹出"
|
||
"浏览器里登录抖音一次；cookie 持久化后自动恢复。",
|
||
"scenario_tags": ["super_agent", "escalation",
|
||
"抖音登录"],
|
||
"target_entity_types": ["Place"], "target_fields": [],
|
||
"suggested_collection_method": "manual_login",
|
||
"priority": 1})
|
||
uid = await get_admin_user_id()
|
||
if uid:
|
||
await create_notification(
|
||
uid, title="抖音需一次性登录",
|
||
body="运行 scripts/douyin_login.py 登录一次即可，"
|
||
"馆长自动恢复。",
|
||
ntype="task", related_task_id=task["id"])
|
||
except Exception:
|
||
pass
|
||
await sa_add_task(run_id, step, "抖音", "douyin",
|
||
"抖音未登录，已升级人工一次性登录", "notify_admin",
|
||
result={"fetched": 0, "approved": 0,
|
||
"pending": 0, "skipped": 0},
|
||
status="escalated",
|
||
note="待 scripts/douyin_login.py")
|
||
await sa_append_step(run_id, {
|
||
"step": step, "action": "douyin",
|
||
"reason": "抖音未登录 → 已开工单+通知(运行 douyin_login.py)，"
|
||
"本轮暂停抖音"})
|
||
return False
|
||
ev_total += int(r.get("evidence_saved") or 0)
|
||
if r.get("found"):
|
||
_apply_dy(t["eid"], r.get("tags") or [])
|
||
found += 1
|
||
if r.get("tags"):
|
||
done += 1
|
||
tags_total += len(r["tags"])
|
||
else:
|
||
_apply_dy(t["eid"], [])
|
||
res_obj = {"fetched": len(targets), "approved": tags_total,
|
||
"pending": ev_total, "skipped": len(targets) - found}
|
||
await sa_add_task(run_id, step, "抖音", "douyin",
|
||
f"抖音 UGC 采集 {len(targets)} 个实体（{last}）",
|
||
"douyin_agent", result=res_obj, status="done",
|
||
note=f"命中{found}·证据{ev_total}条·体验标签{tags_total}")
|
||
await sa_append_step(run_id, {
|
||
"step": step, "action": "douyin",
|
||
"reason": f"抖音采集：命中 {found}/{len(targets)}，证据入库 {ev_total} 条，"
|
||
f"产出 {tags_total} 个体验标签（{last}）"})
|
||
return True
|
||
|
||
|
||
def _events_targets(limit: int) -> list[dict]:
|
||
"""已采过任何源(web/xhs/dy) 但还没抽过事件(无 events_done)的 Place。
|
||
|
||
平台无关：只要 social_evidence 里有该 pnk 的证据(任意 platform)，
|
||
都可以挖事件,不再硬要求 xhs_done。
|
||
"""
|
||
g = _get_graph()
|
||
try:
|
||
rs = g.query(
|
||
"MATCH (p:Place) "
|
||
"WHERE (p.web_done=1 OR p.xhs_done=1 OR p.dy_done=1) "
|
||
" AND p.events_done IS NULL "
|
||
"RETURN p.element_id, p.name LIMIT $n", {"n": limit}).result_set
|
||
except Exception:
|
||
return []
|
||
return [{"eid": r[0], "name": r[1]} for r in rs if r and r[1]]
|
||
|
||
|
||
def _apply_events(eid: str, events: list[dict]) -> None:
|
||
"""事件写回：MERGE Event + (Place)-[:HAS_EVENT{time,type}]->(Event)。
|
||
|
||
- source 不再硬编 'xiaohongshu',按事件实际来源(baike/wiki/xhs/douyin)写
|
||
- 加 time_norm(排序用) / participants(涉及人物) / confidence(置信度)
|
||
- MERGE 用 (place,title) 幂等,同名事件不会重复
|
||
- 无论有无都打 events_done 防重复整批
|
||
"""
|
||
g = _get_graph()
|
||
try:
|
||
g.query("MATCH (p:Place {element_id:$eid}) SET p.events_done=1",
|
||
{"eid": eid})
|
||
for e in events:
|
||
g.query(
|
||
"MATCH (p:Place {element_id:$eid}) "
|
||
"MERGE (ev:Event {place:$eid, title:$t}) "
|
||
"SET ev.time=$tm, ev.time_norm=$tn, ev.type=$ty, "
|
||
" ev.desc=$d, ev.source=$src, ev.participants=$ppl, "
|
||
" ev.confidence=$conf "
|
||
"MERGE (p)-[:HAS_EVENT {time:$tn, type:$ty}]->(ev)",
|
||
{"eid": eid, "t": e.get("title", ""),
|
||
"tm": e.get("time", ""),
|
||
"tn": e.get("time_norm", "") or e.get("time", ""),
|
||
"ty": e.get("type", ""), "d": e.get("desc", ""),
|
||
"src": e.get("source_platform", "") or "mixed",
|
||
"ppl": ",".join(e.get("participants") or []),
|
||
"conf": float(e.get("confidence") or 0)})
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
async def _event_mine(run_id: int, step: int, targets: list[dict]) -> bool:
|
||
"""工具：从证据层评论/帖子抽时间锚定事件(纯 LLM，不爬网)。"""
|
||
done = ev_total = 0
|
||
last = ""
|
||
for t in targets:
|
||
r = await mine_events(t)
|
||
last = r.get("summary", "")
|
||
if not r.get("ok"):
|
||
await sa_add_task(run_id, step, "事件", "event",
|
||
f"event_miner 未就绪：{last}", "event_miner",
|
||
result={"fetched": 0, "approved": 0,
|
||
"pending": 0, "skipped": 0},
|
||
status="skipped", note=last)
|
||
await sa_append_step(run_id, {"step": step, "action": "event",
|
||
"reason": f"事件抽取未配置（{last}）"})
|
||
return False
|
||
evs = r.get("events") or []
|
||
_apply_events(t["eid"], evs)
|
||
if evs:
|
||
done += 1
|
||
ev_total += len(evs)
|
||
res_obj = {"fetched": len(targets), "approved": ev_total,
|
||
"pending": 0, "skipped": len(targets) - done}
|
||
await sa_add_task(run_id, step, "事件", "event",
|
||
f"评论时间→事件抽取 {len(targets)} 个实体（{last}）",
|
||
"event_miner", result=res_obj, status="done",
|
||
note=f"命中{done}·事件{ev_total}")
|
||
await sa_append_step(run_id, {
|
||
"step": step, "action": "event",
|
||
"reason": f"事件抽取：{done}/{len(targets)} 实体，"
|
||
f"产出 {ev_total} 个时间锚定事件（{last}）"})
|
||
return True
|
||
|
||
|
||
async def run_super_agent(run_id: int) -> None:
|
||
await _seed_if_needed()
|
||
step = 0
|
||
api_calls = 0
|
||
enrich_off = False # 记忆蒸馏未配置/不可用时本轮关闭
|
||
web_off = False # web_agent 未就绪时本轮关闭，避免空转
|
||
xhs_off = False # 小红书未配置/未登录时本轮关闭
|
||
dy_off = False # 抖音未配置/未登录时本轮关闭
|
||
event_off = False # 事件抽取未配置时本轮关闭
|
||
enrich_turn = 0 # 富集源轮转(web/xhs/douyin/event/distill 公平)
|
||
escalated: set[str] = set()
|
||
|
||
async def _wait_steward() -> bool:
|
||
waited = 0
|
||
while waited < _STEWARD_INTERVAL:
|
||
if await sa_stop_requested(run_id):
|
||
return True
|
||
await asyncio.sleep(_STEWARD_TICK)
|
||
waited += _STEWARD_TICK
|
||
return False
|
||
|
||
try:
|
||
while True:
|
||
if await sa_stop_requested(run_id):
|
||
await sa_finish(run_id, "stopped")
|
||
return
|
||
|
||
# ── 指挥大脑选工具 ───────────────────────────────────────
|
||
# 高德网格法 = 主力快采（最快、结构化、带坐标）；
|
||
# 多模型蒸馏 = 知识富集（补高德给不了的知识层，独立来源）。
|
||
# 网格忙时每 _ENRICH_EVERY 步插一次富集；网格扫完后全力富集。
|
||
pend = await grid_pending_cats()
|
||
grid_busy = (bool(pend) and api_calls < _MAX_API_PER_RUN
|
||
and step < _HARD_MAX_STEPS)
|
||
|
||
# 富集插槽：web/小红书/抖音/事件/记忆蒸馏 **轮转调度**
|
||
# (避免某源目标海量把其它源饿死，多源公平持续推进)
|
||
if ((not web_off or not xhs_off or not dy_off or not event_off
|
||
or not enrich_off)
|
||
and ((not grid_busy)
|
||
or (step > 0 and step % _ENRICH_EVERY == 0))):
|
||
pool = ([("web", _web_targets, _WEB_BATCH, _web_enrich)]
|
||
if not web_off else [])
|
||
pool += ([("xhs", _xhs_targets, _XHS_BATCH, _xhs_enrich)]
|
||
if not xhs_off else [])
|
||
pool += ([("douyin", _dy_targets, 1, _douyin_enrich)]
|
||
if not dy_off else [])
|
||
pool += ([("event", _events_targets, 3, _event_mine)]
|
||
if not event_off else [])
|
||
pool += ([("distill", _enrich_targets, _ENRICH_BATCH,
|
||
_distill_enrich)] if not enrich_off else [])
|
||
hit = False
|
||
for i in range(len(pool)):
|
||
name, tfn, batch, efn = pool[(enrich_turn + i) % len(pool)]
|
||
tg = await asyncio.to_thread(tfn, batch)
|
||
if not tg:
|
||
continue
|
||
enrich_turn += 1
|
||
step += 1
|
||
ok = await efn(run_id, step, tg)
|
||
if not ok: # 未就绪 → 本轮关该源
|
||
if name == "web":
|
||
web_off = True
|
||
elif name == "xhs":
|
||
xhs_off = True
|
||
elif name == "douyin":
|
||
dy_off = True
|
||
elif name == "event":
|
||
event_off = True
|
||
else:
|
||
enrich_off = True
|
||
break
|
||
hit = True
|
||
break
|
||
if hit:
|
||
await asyncio.sleep(_API_PACING)
|
||
continue
|
||
|
||
if not grid_busy:
|
||
# 网格扫完且无可富集 → 驻守（不烧额度，下轮自动续，可一键停止）
|
||
reason = ("全城网格已扫完、知识富集也已补全，进入驻守巡检"
|
||
"（不再消耗额度，新数据/工单结清后自动恢复，可停止）"
|
||
if not pend else
|
||
"达单进程安全上限，转驻守；下次启动自动从断点续扫")
|
||
await sa_set_status(run_id, "stewarding")
|
||
await sa_append_step(run_id, {"step": step,
|
||
"action": "steward",
|
||
"reason": reason})
|
||
if await _wait_steward():
|
||
await sa_finish(run_id, "stopped")
|
||
return
|
||
api_calls = 0 # 新驻守周期重置进程内计数
|
||
continue
|
||
|
||
# 聚焦：仍有待扫格的类里，挑当前藏品最少的（均衡推进）
|
||
cov = {i["cat"]: i["current"] for i in _coverage()["items"]}
|
||
focus = min(pend, key=lambda c: cov.get(c, 0))
|
||
cell = await grid_take_next(focus)
|
||
if not cell:
|
||
continue
|
||
|
||
tc = cell["typecode"]
|
||
page = cell["next_page"]
|
||
bbox = (cell["min_lng"], cell["min_lat"],
|
||
cell["max_lng"], cell["max_lat"])
|
||
step += 1
|
||
try:
|
||
rows, raw = await asyncio.to_thread(
|
||
search_polygon, tc, bbox, page, _OFFSET)
|
||
except Exception as e:
|
||
rows, raw = [], -1
|
||
err = str(e)[:140]
|
||
else:
|
||
err = None
|
||
api_calls += 1
|
||
|
||
res = await ingest_rows(rows, focus, focus) if rows else \
|
||
{"fetched": 0, "approved": 0, "pending": 0, "skipped": 0}
|
||
if err:
|
||
res["error"] = err
|
||
|
||
area = (f"经[{bbox[0]:.3f},{bbox[2]:.3f}] "
|
||
f"纬[{bbox[1]:.3f},{bbox[3]:.3f}]")
|
||
# 状态机：扫尽 / 翻页 / 自适应细分(密集区 8 页仍满→四叉细分,防 POI 丢)
|
||
if err:
|
||
# 单格请求出错：不前进、不标尽，留待后续重试（不浪费已得）
|
||
nstatus, npage, note = "pending", page, f"请求失败重试 {err}"
|
||
elif raw <= 0 or raw < _OFFSET:
|
||
nstatus, npage, note = "exhausted", page, f"网格扫尽(本页{raw}条)"
|
||
elif page >= _MAX_PAGE_SPLIT and cell["depth"] < _MAX_DEPTH:
|
||
await grid_subdivide(cell["id"], focus, tc,
|
||
_quads(cell), cell["depth"] + 1)
|
||
nstatus, npage, note = None, page, "稠密→四叉细分深扫"
|
||
elif page >= _MAX_PAGE_SPLIT:
|
||
nstatus, npage, note = "exhausted", page, "达最深层，止于该格"
|
||
else:
|
||
nstatus, npage, note = "pending", page + 1, "本页满，续下一页"
|
||
|
||
if nstatus is not None:
|
||
await grid_update(cell["id"], npage,
|
||
res.get("approved", 0) + res.get("pending", 0),
|
||
nstatus)
|
||
|
||
reason = f"全城网格扫描 {area} 第{page}页（{focus}/{tc}）"
|
||
await sa_add_task(run_id, step, focus, "ingest", reason,
|
||
"gaode_grid", result=res, status="done",
|
||
note=f"格#{cell['id']} d{cell['depth']} p{page}·{note}")
|
||
await sa_append_step(run_id, {
|
||
"step": step,
|
||
"plan": {"poi_type": focus, "keyword": f"网格 {area} p{page}",
|
||
"reason": reason},
|
||
"result": res,
|
||
}, ingested_delta=res.get("approved", 0))
|
||
|
||
# 该类全城扫尽却仍偏少 → 求助管理员（去重，馆长不停）
|
||
if focus not in escalated:
|
||
gc = await grid_counts()
|
||
g = gc.get(focus, {})
|
||
if g.get("total") and g["done"] >= g["total"]:
|
||
maxc = max(cov.values()) if cov else 0
|
||
if cov.get(focus, 0) < max(0.25 * maxc, 30):
|
||
await _escalate(run_id, step, focus, cov.get(focus, 0))
|
||
escalated.add(focus)
|
||
|
||
await asyncio.sleep(_API_PACING)
|
||
except Exception as e: # noqa: BLE001
|
||
await sa_finish(run_id, "error", str(e)[:240])
|
||
|
||
|
||
def schedule_super_agent(run_id: int) -> None:
|
||
t = asyncio.create_task(run_super_agent(run_id))
|
||
_BG.add(t)
|
||
t.add_done_callback(_BG.discard)
|