|
|
@@ -1,422 +0,0 @@
|
|
|
-"""
|
|
|
-在流程结束后写入**内容策略表格** JSON，并回写 MySQL。
|
|
|
-
|
|
|
-输出路径：{OUTPUT_DIR}/{trace_id}/process_trace.json
|
|
|
-每条策略行另按 (trace_id, aweme_id) 更新 demand_find_content_result.process_trace（TEXT）。
|
|
|
-"""
|
|
|
-
|
|
|
-from __future__ import annotations
|
|
|
-
|
|
|
-import json
|
|
|
-import logging
|
|
|
-import os
|
|
|
-from pathlib import Path
|
|
|
-from typing import Any, Dict, List, Optional, Tuple
|
|
|
-
|
|
|
-from agent.tools import tool, ToolResult
|
|
|
-from utils.tool_logging import format_tool_result_for_log, log_tool_call
|
|
|
-
|
|
|
-from db import update_process_trace_by_aweme_id
|
|
|
-
|
|
|
-_LOG_LABEL = "工具调用：exec_summary -> 写入过程 trace JSON"
|
|
|
-
|
|
|
-logger = logging.getLogger(__name__)
|
|
|
-
|
|
|
-
|
|
|
-def _output_dir_path() -> Path:
|
|
|
- # 与 store_results_mysql / output.json 目录约定一致
|
|
|
- return Path(os.getenv("OUTPUT_DIR", ".cache/output"))
|
|
|
-
|
|
|
-
|
|
|
-def _parse_payload(summary_json: str) -> Dict[str, Any]:
|
|
|
- """
|
|
|
- 解析并规范化 LLM 传入的表格数据。
|
|
|
-
|
|
|
- - 如果是数组：视为“表格行列表”，包成 {"rows": [...]}
|
|
|
- - 如果是对象：直接返回（用于后续扩展字段）
|
|
|
- """
|
|
|
- data = json.loads(summary_json)
|
|
|
- if isinstance(data, list):
|
|
|
- return {"rows": data}
|
|
|
- if not isinstance(data, dict):
|
|
|
- raise ValueError("summary_json 解析后必须是 JSON 对象或数组")
|
|
|
- return data
|
|
|
-
|
|
|
-
|
|
|
-def _split_input_features(raw: str) -> List[str]:
|
|
|
- s = (raw or "").strip()
|
|
|
- if not s:
|
|
|
- return []
|
|
|
- parts = s.replace("，", ",").split(",")
|
|
|
- out: List[str] = []
|
|
|
- for p in parts:
|
|
|
- t = p.strip()
|
|
|
- if t:
|
|
|
- out.append(t)
|
|
|
- return out
|
|
|
-
|
|
|
-
|
|
|
-def _load_output_json(*, trace_id: str) -> Optional[Dict[str, Any]]:
|
|
|
- path = _output_dir_path() / trace_id / "output.json"
|
|
|
- try:
|
|
|
- with path.open("r", encoding="utf-8") as f:
|
|
|
- data = json.load(f)
|
|
|
- except FileNotFoundError:
|
|
|
- return None
|
|
|
- except Exception:
|
|
|
- logger.warning("读取 output.json 失败: %s", str(path), exc_info=True)
|
|
|
- return None
|
|
|
- return data if isinstance(data, dict) else None
|
|
|
-
|
|
|
-
|
|
|
-def _extract_contents(*, trace_id: str) -> List[Dict[str, Any]]:
|
|
|
- """
|
|
|
- 从 output.json 读取最终入选 contents。
|
|
|
-
|
|
|
- 约定：
|
|
|
- - 只允许对 output.json.contents 内的 aweme_id 生成/写入 process_trace rows
|
|
|
- """
|
|
|
- output_json = _load_output_json(trace_id=trace_id) or {}
|
|
|
- contents = output_json.get("contents")
|
|
|
- if not isinstance(contents, list):
|
|
|
- return []
|
|
|
- out: List[Dict[str, Any]] = []
|
|
|
- for item in contents:
|
|
|
- if isinstance(item, dict):
|
|
|
- out.append(item)
|
|
|
- return out
|
|
|
-
|
|
|
-
|
|
|
-def _map_strategy_type(value: Any) -> str:
|
|
|
- v = str(value or "").strip()
|
|
|
- if v in ("case_based", "case", "case出发"):
|
|
|
- return "case出发"
|
|
|
- if v in ("feature_based", "feature", "特征出发"):
|
|
|
- return "特征出发"
|
|
|
- return v
|
|
|
-
|
|
|
-
|
|
|
-def _map_channel(value: Any) -> str:
|
|
|
- v = str(value or "").strip()
|
|
|
- mapping = {
|
|
|
- "search": "抖音搜索",
|
|
|
- "author": "订阅账号",
|
|
|
- "ranking": "榜单",
|
|
|
- "other": "其他",
|
|
|
- "抖音搜索": "抖音搜索",
|
|
|
- "订阅账号": "订阅账号",
|
|
|
- "榜单": "榜单",
|
|
|
- "其他": "其他",
|
|
|
- }
|
|
|
- return mapping.get(v, v)
|
|
|
-
|
|
|
-
|
|
|
-def _map_decision_basis(value: Any) -> str:
|
|
|
- v = str(value or "").strip()
|
|
|
- mapping = {
|
|
|
- "content_portrait": "内容画像匹配",
|
|
|
- "author_portrait": "作者画像匹配",
|
|
|
- "demand_filtering": "需求筛选",
|
|
|
- "other": "其他",
|
|
|
- "画像缺失": "画像缺失",
|
|
|
- "内容画像匹配": "内容画像匹配",
|
|
|
- "作者画像匹配": "作者画像匹配",
|
|
|
- "需求筛选": "需求筛选",
|
|
|
- "其他": "其他",
|
|
|
- }
|
|
|
- return mapping.get(v, v)
|
|
|
-
|
|
|
-
|
|
|
-def _infer_decision_basis_from_output_content(content: Dict[str, Any]) -> str:
|
|
|
- portrait = content.get("portrait_data") or {}
|
|
|
- source = str(portrait.get("source") or "").strip()
|
|
|
- if source == "content_like":
|
|
|
- return "内容画像匹配"
|
|
|
- if source == "account_fans":
|
|
|
- return "作者画像匹配"
|
|
|
- if source == "none":
|
|
|
- return "画像缺失"
|
|
|
- return ""
|
|
|
-
|
|
|
-
|
|
|
-def _build_base_row(*, trace_id: str, content: Dict[str, Any], input_features: List[str], query: str) -> Dict[str, Any]:
|
|
|
- return {
|
|
|
- "trace_id": trace_id,
|
|
|
- "aweme_id": str(content.get("aweme_id") or "").strip(),
|
|
|
- "title": str(content.get("title") or "").strip(),
|
|
|
- "author_nickname": str(content.get("author_nickname") or "").strip(),
|
|
|
- "strategy_type": "",
|
|
|
- "from_case_aweme_id": "",
|
|
|
- "from_case_point": "",
|
|
|
- "from_feature": "",
|
|
|
- "search_keyword": str(query or "").strip(),
|
|
|
- "channel": "抖音搜索",
|
|
|
- "decision_basis": _infer_decision_basis_from_output_content(content),
|
|
|
- "decision_notes": str(content.get("reason") or "").strip(),
|
|
|
- "input_features": input_features,
|
|
|
- }
|
|
|
-
|
|
|
-
|
|
|
-_ROW_KEYS: Tuple[str, ...] = (
|
|
|
- "trace_id",
|
|
|
- "aweme_id",
|
|
|
- "title",
|
|
|
- "author_nickname",
|
|
|
- "strategy_type",
|
|
|
- "from_case_aweme_id",
|
|
|
- "from_case_point",
|
|
|
- "from_feature",
|
|
|
- "search_keyword",
|
|
|
- "channel",
|
|
|
- "decision_basis",
|
|
|
- "decision_notes",
|
|
|
- "input_features",
|
|
|
-)
|
|
|
-
|
|
|
-
|
|
|
-def _sanitize_row(row: Dict[str, Any]) -> Dict[str, Any]:
|
|
|
- """只保留固定字段，并把枚举值规范成中文。"""
|
|
|
- out: Dict[str, Any] = {k: row.get(k, "") for k in _ROW_KEYS}
|
|
|
- out["strategy_type"] = _map_strategy_type(out.get("strategy_type"))
|
|
|
- out["channel"] = _map_channel(out.get("channel"))
|
|
|
- out["decision_basis"] = _map_decision_basis(out.get("decision_basis"))
|
|
|
- # input_features 规范为 list[str]
|
|
|
- feats = out.get("input_features")
|
|
|
- if isinstance(feats, list):
|
|
|
- out["input_features"] = [str(x).strip() for x in feats if str(x).strip()]
|
|
|
- elif isinstance(feats, str):
|
|
|
- out["input_features"] = _split_input_features(feats)
|
|
|
- else:
|
|
|
- out["input_features"] = []
|
|
|
- return out
|
|
|
-
|
|
|
-
|
|
|
-def _normalize_payload(*, trace_id: str, payload: Dict[str, Any]) -> Dict[str, Any]:
|
|
|
- # tool 只做最小职责：过滤/补全/规范化；复杂推理由 skill 生成 summary_json 来完成
|
|
|
- raw_rows = payload.get("rows")
|
|
|
- rows_in_payload: List[Dict[str, Any]] = []
|
|
|
- if isinstance(raw_rows, list):
|
|
|
- for item in raw_rows:
|
|
|
- if isinstance(item, dict):
|
|
|
- rows_in_payload.append(item)
|
|
|
-
|
|
|
- output_json = _load_output_json(trace_id=trace_id) or {}
|
|
|
- query = str(output_json.get("query") or "").strip()
|
|
|
- input_features = _split_input_features(query)
|
|
|
- contents = _extract_contents(trace_id=trace_id)
|
|
|
- contents_by_aweme_id: Dict[str, Dict[str, Any]] = {
|
|
|
- str(c.get("aweme_id") or "").strip(): c for c in contents if str(c.get("aweme_id") or "").strip()
|
|
|
- }
|
|
|
-
|
|
|
- # 先把 payload rows 归并到 aweme_id
|
|
|
- payload_by_aweme_id: Dict[str, Dict[str, Any]] = {}
|
|
|
- for r in rows_in_payload:
|
|
|
- aweme_id = str(r.get("aweme_id") or r.get("awemeId") or "").strip()
|
|
|
- if not aweme_id:
|
|
|
- continue
|
|
|
- payload_by_aweme_id[aweme_id] = dict(r)
|
|
|
-
|
|
|
- # 只允许 payload 覆盖“策略/来源/解释”字段，避免覆盖 output.json.contents 的身份字段（title/author 等）
|
|
|
- allowed_payload_keys: set[str] = {
|
|
|
- "strategy_type",
|
|
|
- "from_case_aweme_id",
|
|
|
- "from_case_point",
|
|
|
- "from_feature",
|
|
|
- "search_keyword",
|
|
|
- "channel",
|
|
|
- "decision_basis",
|
|
|
- "decision_notes",
|
|
|
- "input_features",
|
|
|
- }
|
|
|
-
|
|
|
- # 兼容 payload 的常见别名/驼峰 key（模型输出不稳定时，尽量不丢信息）
|
|
|
- alias_map: Dict[str, Tuple[str, ...]] = {
|
|
|
- "strategy_type": ("strategy_type", "strategyType"),
|
|
|
- "from_case_aweme_id": ("from_case_aweme_id", "fromCaseAwemeId", "case_aweme_id", "caseAwemeId"),
|
|
|
- "from_case_point": ("from_case_point", "fromCasePoint", "case_point", "casePoint"),
|
|
|
- "from_feature": ("from_feature", "fromFeature", "feature", "from_feature_name"),
|
|
|
- "search_keyword": ("search_keyword", "searchKeyword", "keyword"),
|
|
|
- "channel": ("channel", "source_channel", "sourceChannel", "source"),
|
|
|
- "decision_basis": ("decision_basis", "decisionBasis"),
|
|
|
- "decision_notes": ("decision_notes", "decisionNotes", "notes"),
|
|
|
- "input_features": ("input_features", "inputFeatures"),
|
|
|
- }
|
|
|
-
|
|
|
- def _pick(provided: Dict[str, Any], key: str) -> Any:
|
|
|
- for k in alias_map.get(key, (key,)):
|
|
|
- if k in provided:
|
|
|
- return provided.get(k)
|
|
|
- return None
|
|
|
-
|
|
|
- normalized: List[Dict[str, Any]] = []
|
|
|
- for aweme_id, content in contents_by_aweme_id.items():
|
|
|
- base = _build_base_row(trace_id=trace_id, content=content, input_features=input_features, query=query)
|
|
|
- provided = payload_by_aweme_id.get(aweme_id) or {}
|
|
|
-
|
|
|
- merged = dict(base)
|
|
|
- # 只合并允许覆盖的字段
|
|
|
- for k in allowed_payload_keys:
|
|
|
- v = _pick(provided, k)
|
|
|
- if v is not None:
|
|
|
- merged[k] = v
|
|
|
-
|
|
|
- # 身份字段强制以 output.json.contents 为准（即使 payload 传了也不采纳）
|
|
|
- merged["aweme_id"] = str(content.get("aweme_id") or "").strip()
|
|
|
- merged["title"] = str(content.get("title") or "").strip()
|
|
|
- merged["author_nickname"] = str(content.get("author_nickname") or "").strip()
|
|
|
-
|
|
|
- # 如果缺失 input_features，用 query 拆分补齐
|
|
|
- if "input_features" not in merged or not merged.get("input_features"):
|
|
|
- merged["input_features"] = input_features
|
|
|
-
|
|
|
- normalized.append(_sanitize_row(merged))
|
|
|
-
|
|
|
- # 保持稳定顺序：按 rank（若有）或 aweme_id
|
|
|
- def _sort_key(r: Dict[str, Any]) -> Tuple[int, str]:
|
|
|
- c = contents_by_aweme_id.get(str(r.get("aweme_id") or "").strip()) or {}
|
|
|
- try:
|
|
|
- rank = int(c.get("rank") or 0)
|
|
|
- except Exception:
|
|
|
- rank = 0
|
|
|
- return (rank if rank > 0 else 10**9, str(r.get("aweme_id") or ""))
|
|
|
-
|
|
|
- normalized.sort(key=_sort_key)
|
|
|
- return {"rows": normalized}
|
|
|
-
|
|
|
-
|
|
|
-def _write_process_trace(*, trace_id: str, payload: Dict[str, Any]) -> Path:
|
|
|
- out_dir = _output_dir_path() / trace_id
|
|
|
- out_dir.mkdir(parents=True, exist_ok=True)
|
|
|
- path = out_dir / "process_trace.json"
|
|
|
- # 输出格式收敛：只允许 {"rows": [...]}
|
|
|
- doc = {"rows": payload.get("rows") or []}
|
|
|
- with path.open("w", encoding="utf-8") as f:
|
|
|
- json.dump(doc, f, ensure_ascii=False, indent=2)
|
|
|
- return path
|
|
|
-
|
|
|
-
|
|
|
-def _sync_process_trace_rows_to_mysql(*, trace_id: str, rows: List[Dict[str, Any]]) -> Dict[str, Any]:
|
|
|
- """
|
|
|
- 将每条归一化后的策略行序列化为 JSON 文本，按 (trace_id, aweme_id) 更新 process_trace 与 channel。
|
|
|
-
|
|
|
- channel 当前统一为「抖音」（与 process_trace.json 内 channel「抖音搜索」区分）。
|
|
|
- 表中无匹配行时 rowcount 为 0，计入 skipped。
|
|
|
- """
|
|
|
- updated = 0
|
|
|
- skipped = 0
|
|
|
- errors: List[str] = []
|
|
|
- for row in rows:
|
|
|
- aweme_id = str(row.get("aweme_id") or "").strip()
|
|
|
- if not aweme_id:
|
|
|
- skipped += 1
|
|
|
- continue
|
|
|
- text = json.dumps(row, ensure_ascii=False)
|
|
|
- try:
|
|
|
- n = update_process_trace_by_aweme_id(
|
|
|
- trace_id=trace_id,
|
|
|
- aweme_id=aweme_id,
|
|
|
- process_trace_text=text,
|
|
|
- channel="抖音",
|
|
|
- )
|
|
|
- if n > 0:
|
|
|
- updated += 1
|
|
|
- else:
|
|
|
- skipped += 1
|
|
|
- except Exception as e:
|
|
|
- logger.warning(
|
|
|
- "process_trace 回写 MySQL 失败 trace_id=%s aweme_id=%s: %s",
|
|
|
- trace_id,
|
|
|
- aweme_id,
|
|
|
- e,
|
|
|
- exc_info=True,
|
|
|
- )
|
|
|
- errors.append(f"{aweme_id}: {e}")
|
|
|
- return {"updated": updated, "skipped": skipped, "errors": errors}
|
|
|
-
|
|
|
-
|
|
|
-@tool(
|
|
|
- description=(
|
|
|
- "在**全部流程执行完毕之后**调用：把每条最终入选内容的「选择策略」整理成表格 JSON，"
|
|
|
- "写入当前任务的 output 目录下的 process_trace.json，便于后续复盘；"
|
|
|
- "并将每一行策略 JSON 序列化为文本，按 trace_id + aweme_id 回写到 "
|
|
|
- "demand_find_content_result.process_trace，并同步将 channel 字段设为「抖音」。"
|
|
|
- "参数 summary_json 为 JSON 字符串，可以是数组或对象（对象需包含 rows）。"
|
|
|
- "可选参数 log_path/log_text 用于传入本次运行日志（便于复盘留档/未来扩展）。"
|
|
|
- ),
|
|
|
-)
|
|
|
-async def exec_summary(
|
|
|
- trace_id: str,
|
|
|
- summary_json: str,
|
|
|
- log_path: str = "",
|
|
|
- log_text: str = "",
|
|
|
-) -> ToolResult:
|
|
|
- call_params = {
|
|
|
- "trace_id": trace_id,
|
|
|
- "summary_json": "<json>",
|
|
|
- "log_path": (log_path or "").strip(),
|
|
|
- "log_text": "<text>",
|
|
|
- }
|
|
|
- tid = (trace_id or "").strip()
|
|
|
- if not tid:
|
|
|
- err = ToolResult(
|
|
|
- title="过程摘要",
|
|
|
- output="trace_id 不能为空",
|
|
|
- metadata={"ok": False, "error": "empty trace_id"},
|
|
|
- )
|
|
|
- log_tool_call(_LOG_LABEL, call_params, format_tool_result_for_log(err))
|
|
|
- return err
|
|
|
-
|
|
|
- try:
|
|
|
- payload = _parse_payload(summary_json)
|
|
|
- except json.JSONDecodeError as e:
|
|
|
- err = ToolResult(
|
|
|
- title="过程摘要",
|
|
|
- output=f"summary_json 不是合法 JSON: {e}",
|
|
|
- metadata={"ok": False, "error": str(e)},
|
|
|
- )
|
|
|
- log_tool_call(_LOG_LABEL, call_params, format_tool_result_for_log(err))
|
|
|
- return err
|
|
|
- except ValueError as e:
|
|
|
- err = ToolResult(
|
|
|
- title="过程摘要",
|
|
|
- output=str(e),
|
|
|
- metadata={"ok": False, "error": str(e)},
|
|
|
- )
|
|
|
- log_tool_call(_LOG_LABEL, call_params, format_tool_result_for_log(err))
|
|
|
- return err
|
|
|
-
|
|
|
- payload = _normalize_payload(trace_id=tid, payload=payload)
|
|
|
-
|
|
|
- try:
|
|
|
- path = _write_process_trace(trace_id=tid, payload=payload)
|
|
|
- except OSError as e:
|
|
|
- msg = f"写入 process_trace.json 失败: {e}"
|
|
|
- logger.error(msg, exc_info=True)
|
|
|
- err = ToolResult(title="过程摘要", output=msg, metadata={"ok": False, "error": str(e)})
|
|
|
- log_tool_call(_LOG_LABEL, call_params, format_tool_result_for_log(err))
|
|
|
- return err
|
|
|
-
|
|
|
- rows = payload.get("rows") or []
|
|
|
- mysql_meta: Dict[str, Any]
|
|
|
- try:
|
|
|
- mysql_meta = _sync_process_trace_rows_to_mysql(trace_id=tid, rows=rows if isinstance(rows, list) else [])
|
|
|
- except Exception as e:
|
|
|
- logger.warning("process_trace 批量回写 MySQL 异常: %s", e, exc_info=True)
|
|
|
- mysql_meta = {"updated": 0, "skipped": 0, "errors": [str(e)]}
|
|
|
-
|
|
|
- out = ToolResult(
|
|
|
- title="过程摘要",
|
|
|
- output=f"已写入 {path}；MySQL process_trace 已更新 {mysql_meta.get('updated', 0)} 条",
|
|
|
- metadata={
|
|
|
- "ok": True,
|
|
|
- "trace_id": tid,
|
|
|
- "path": str(path),
|
|
|
- "log_path": (log_path or "").strip(),
|
|
|
- "log_text_len": len((log_text or "").strip()),
|
|
|
- "mysql_process_trace_updated": mysql_meta.get("updated", 0),
|
|
|
- "mysql_process_trace_skipped": mysql_meta.get("skipped", 0),
|
|
|
- "mysql_process_trace_errors": mysql_meta.get("errors") or [],
|
|
|
- },
|
|
|
- )
|
|
|
- log_tool_call(_LOG_LABEL, {"trace_id": tid}, format_tool_result_for_log(out))
|
|
|
- return out
|
