#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 第一步分析脚本 v2 基于过滤后的 how 解构结果，分析哪些点最有可能是创作者的第一步（创作起点）， 以及第一步的前序节点（第一步是怎么来的）。 v2 新增功能： - 分析第一步的前序节点（已有人设 或 外部触发） - 提供搜索关键词（用于后续验证） - 严格约束：前序节点只能是已匹配的人设节点，搜索关键词只能来自节点名称 输入：intermediate/filtered_results/ 中的过滤结果 输出：第一步分析结果（带前序节点信息） """ import asyncio import json from pathlib import Path from typing import Dict, List import sys # 添加项目根目录到路径 project_root = Path(__file__).parent.parent.parent sys.path.insert(0, str(project_root)) from agents import Agent, Runner, ModelSettings, trace from agents.tracing.create import custom_span from lib.client import get_model from lib.my_trace import set_trace_smith as set_trace from script.data_processing.path_config import PathConfig # 模型配置 MODEL_NAME = "google/gemini-3-pro-preview" # MODEL_NAME = 'anthropic/claude-sonnet-4.5' agent = Agent( name="First Step Analyzer V2", model=get_model(MODEL_NAME), model_settings=ModelSettings( temperature=0.0, max_tokens=65536, ), tools=[], ) def extract_points_from_filtered_result(how_result: Dict) -> List[Dict]: """ 从过滤后的 how 解构结果中提取所有点的信息 Args: how_result: how解构结果 Returns: 点信息列表，不包含原始分类标签 """ points = [] for point_type in ["灵感点", "关键点", "目的点"]: point_list_key = f"{point_type}列表" point_list = how_result.get(point_list_key, []) for point in point_list: point_name = point.get("名称", "") point_desc = point.get("描述", "") # 检查是否有匹配到人设特征 has_match = False matched_feature = None similarity = None # 遍历 how 步骤列表中的特征 for step in point.get("how步骤列表", []): for feature in step.get("特征列表", []): match_results = feature.get("匹配结果", []) if match_results: # 如果有匹配结果（top1） has_match = True match = match_results[0] matched_feature = match.get("人设特征名称", "") similarity = match.get("匹配结果", {}).get("相似度", 0) break if has_match: break point_info = { "名称": point_name, "描述": point_desc, "是否匹配到已有人设": has_match } if has_match: point_info["匹配的人设特征"] = matched_feature point_info["相似度"] = similarity points.append(point_info) return points def build_prompt(points: List[Dict]) -> str: """ 构建分析 prompt (v2版本：增加前序节点分析) Args: points: 点信息列表 Returns: prompt 文本 """ # 构建点的描述文本 points_text = [] matched_points_names = [] # 收集已匹配的点名称 for i, point in enumerate(points, 1): text = f"{i}. {point['名称']}\n {point['描述']}" if point['是否匹配到已有人设']: text += f"\n [已匹配] 匹配到人设特征: {point['匹配的人设特征']} (相似度: {point['相似度']:.2f})" matched_points_names.append(point['名称']) else: text += "\n [未匹配] 未匹配到已有人设特征" points_text.append(text) points_section = "\n\n".join(points_text) matched_points_section = "\n".join([f"- {name}" for name in matched_points_names]) return f''' 以下是一个内容创作的解构结果。这些点已经被分析和分类，但这个分类是分析维度，不代表真实的创作顺序。 请判断：在这些点中，哪些最有可能是创作者的"第一步"（创作起点），以及第一步的前序节点（第一步是怎么来的）。 ## 判断标准 **起点特征**： - 最先触发创作、不依赖其他点的节点 - 可能是外部事件、时事热点、商业需求等 - 起点可能有多个 **前序节点规则**： - 每个第一步只有**一个**直接前序节点 - 客观分析哪个前序节点最有可能引发第一步，给出前序概率 - 前序节点只能是以下两种之一： 1. **已匹配的人设节点**：从下面的"已匹配的点"列表中选择 2. **外部触发**：纯外部事件触发 - 选择前序概率最高的那个，不要预设倾向 **搜索关键词规则**（重要！）： - 如果前序是外部触发，必须提供搜索关键词 - 搜索关键词**只能**从点的名称中提取，不能推导、不能扩展、不能添加任何额外的词 ## 已匹配的点（可作为前序节点） {matched_points_section} ## 待分析的点 {points_section} ## 输出要求 以 JSON 格式输出： {{ "推理过程": "详细说明判断逻辑...", "第一步候选": [ {{ "点名称": "...", "第一步概率": 0.95, "推理依据": "...", "来源分析": "外部触发/人设延伸/商业驱动/其他", "前序节点": {{ "类型": "已有人设" 或 "外部触发", "人设节点名称": "..." 或 null, // 如果类型是"已有人设"，必须从上面的"已匹配的点"中选择 "匹配的人设特征": "..." 或 null, // 如果有人设节点，填写其匹配的特征 "相似度": 0.84 或 null, // 人设节点与特征的匹配相似度 "前序概率": 0.75, // 前序节点引发第一步的概率（0-1之间） "搜索关键词": ["关键词1", "关键词2"] 或 null, // 只在"外部触发"时提供，且只能从点名称中提取 "推理": "说明为什么这个前序节点可能引发第一步" }} }} ] }} 注意： 1. 只输出最有可能是第一步的点（通常1-3个） 2. 按第一步概率降序排列 3. 客观分析前序节点，选择前序概率最高的（无论是已匹配人设还是外部触发） 4. 搜索关键词只能从点名称中提取，不能推导 5. 前序节点不需要能完全推导出第一步，只要可能引发创作者关注即可 '''.strip() async def analyze_post(post_data: Dict) -> Dict: """ 分析单个帖子 Args: post_data: 帖子数据（包含 how解构结果） Returns: 分析结果 """ post_id = post_data.get("帖子id", "") how_result = post_data.get("how解构结果", {}) # 提取所有点的信息 points = extract_points_from_filtered_result(how_result) if not points: return { "帖子id": post_id, "模型": MODEL_NAME, "输入": {"点列表": []}, "输出": None, "错误": "没有可分析的点" } # 构建 prompt prompt = build_prompt(points) # 使用 custom_span 标识单个帖子的分析流程 with custom_span( name=f"分析第一步 - 帖子 {post_id}", data={ "帖子id": post_id, "点数量": len(points), "模型": MODEL_NAME } ): # 调用 agent result = await Runner.run(agent, input=prompt) output = result.final_output # 解析 JSON try: if "```json" in output: json_start = output.find("```json") + 7 json_end = output.find("```", json_start) json_str = output[json_start:json_end].strip() elif "{" in output and "}" in output: json_start = output.find("{") json_end = output.rfind("}") + 1 json_str = output[json_start:json_end] else: json_str = output analysis_result = json.loads(json_str) return { "帖子id": post_id, "模型": MODEL_NAME, "输入": { "点列表": points, "prompt": prompt }, "输出": analysis_result } except Exception as e: return { "帖子id": post_id, "模型": MODEL_NAME, "输入": { "点列表": points, "prompt": prompt }, "输出": None, "错误": str(e), "原始输出": output } async def main(current_time: str = None, log_url: str = None): """主函数 Args: current_time: 当前时间戳（从外部传入） log_url: 日志链接（从外部传入） """ # 使用路径配置 config = PathConfig() # 确保输出目录存在 config.ensure_dirs() # 获取路径 input_dir = config.intermediate_dir / "filtered_results" output_dir = config.intermediate_dir / "first_step_analysis_v2" # 确保输出目录存在 output_dir.mkdir(parents=True, exist_ok=True) print(f"账号: {config.account_name}") print(f"输入目录: {input_dir}") print(f"输出目录: {output_dir}") print(f"使用模型: {MODEL_NAME}") if log_url: print(f"Trace URL: {log_url}") print() # 读取所有过滤后的文件 input_files = list(input_dir.glob("*_filtered.json")) if not input_files: print(f"错误: 在 {input_dir} 中没有找到任何 *_filtered.json 文件") return print(f"找到 {len(input_files)} 个文件待分析\n") # 批量分析 results = [] for i, input_file in enumerate(input_files, 1): print(f"[{i}/{len(input_files)}] 分析文件: {input_file.name}") # 读取文件 with open(input_file, "r", encoding="utf-8") as f: post_data = json.load(f) # 分析 result = await analyze_post(post_data) results.append(result) # 立即保存单个帖子的结果 post_id = result.get("帖子id", "unknown") single_output_file = output_dir / f"{post_id}_first_step.json" single_result = { "元数据": { "current_time": current_time, "log_url": log_url, "model": MODEL_NAME }, "帖子id": post_id, "分析结果": result } with open(single_output_file, "w", encoding="utf-8") as f: json.dump(single_result, f, ensure_ascii=False, indent=2) # 显示结果 output = result.get("输出", {}) if output: first_steps = output.get("第一步候选", []) print(f" 第一步候选:") for step in first_steps: print(f" - {step.get('点名称', 'N/A')} (概率: {step.get('第一步概率', 0):.2f})") print(f" ✓ 已保存: {single_output_file.name}") else: print(f" 分析失败: {result.get('错误', 'N/A')}") print() print(f"✓ 所有分析完成，结果已保存到: {output_dir}") if log_url: print(f"Trace: {log_url}") # 打印汇总 print("\n" + "=" * 80) print("分析汇总") print("=" * 80) for result in results: post_id = result["帖子id"] output = result.get("输出", {}) if output: first_steps = output.get("第一步候选", []) print(f"\n帖子 {post_id}:") for step in first_steps: print(f" - {step.get('点名称', 'N/A')} ({step.get('来源分析', 'N/A')}, 概率: {step.get('第一步概率', 0):.2f})") else: print(f"\n帖子 {post_id}: 分析失败") if __name__ == "__main__": # 设置 trace current_time, log_url = set_trace() # 使用 trace 上下文包裹整个执行流程 with trace("第一步分析"): asyncio.run(main(current_time, log_url))