|
|
@@ -0,0 +1,811 @@
|
|
|
+import asyncio
|
|
|
+import json
|
|
|
+import os
|
|
|
+import argparse
|
|
|
+from datetime import datetime
|
|
|
+
|
|
|
+from agents import Agent, Runner
|
|
|
+from lib.my_trace import set_trace
|
|
|
+from typing import Literal
|
|
|
+from pydantic import BaseModel, Field
|
|
|
+
|
|
|
+from lib.utils import read_file_as_string
|
|
|
+from lib.client import get_model
|
|
|
+MODEL_NAME = "google/gemini-2.5-flash"
|
|
|
+from script.search_recommendations.xiaohongshu_search_recommendations import XiaohongshuSearchRecommendations
|
|
|
+
|
|
|
+
|
|
|
+class RunContext(BaseModel):
|
|
|
+ version: str = Field(..., description="当前运行的脚本版本（文件名）")
|
|
|
+ input_files: dict[str, str] = Field(..., description="输入文件路径映射")
|
|
|
+ q_with_context: str
|
|
|
+ q_context: str
|
|
|
+ q: str
|
|
|
+ log_url: str
|
|
|
+ log_dir: str
|
|
|
+
|
|
|
+ # 步骤化日志
|
|
|
+ steps: list[dict] = Field(default_factory=list, description="执行步骤的详细记录")
|
|
|
+
|
|
|
+ # 探索阶段记录（保留用于向后兼容）
|
|
|
+ keywords: list[str] | None = Field(default=None, description="提取的关键词")
|
|
|
+ exploration_levels: list[dict] = Field(default_factory=list, description="每一层的探索结果")
|
|
|
+ level_analyses: list[dict] = Field(default_factory=list, description="每一层的主Agent分析")
|
|
|
+
|
|
|
+ # 最终结果
|
|
|
+ final_candidates: list[str] | None = Field(default=None, description="最终选出的候选query")
|
|
|
+ evaluation_results: list[dict] | None = Field(default=None, description="候选query的评估结果")
|
|
|
+ optimization_result: dict | None = Field(default=None, description="最终优化结果对象")
|
|
|
+ final_output: str | None = Field(default=None, description="最终输出结果（格式化文本）")
|
|
|
+
|
|
|
+
|
|
|
+# ============================================================================
|
|
|
+# Agent 1: 关键词提取专家
|
|
|
+# ============================================================================
|
|
|
+keyword_extraction_instructions = """
|
|
|
+你是关键词提取专家。给定一个搜索问题（含上下文），提取出**最细粒度的关键概念**。
|
|
|
+
|
|
|
+## 提取原则
|
|
|
+
|
|
|
+1. **细粒度优先**：拆分成最小的有意义单元
|
|
|
+ - 不要保留完整的长句
|
|
|
+ - 拆分成独立的、有搜索意义的词或短语
|
|
|
+
|
|
|
+2. **保留核心维度**：
|
|
|
+ - 地域/对象
|
|
|
+ - 时间
|
|
|
+ - 行为/意图：获取、教程、推荐、如何等
|
|
|
+ - 主题/领域
|
|
|
+ - 质量/属性
|
|
|
+
|
|
|
+3. **去掉无意义的虚词**：的、吗、呢等
|
|
|
+
|
|
|
+4. **保留领域专有词**：不要过度拆分专业术语
|
|
|
+ - 如果是常见的组合词，保持完整
|
|
|
+
|
|
|
+## 输出要求
|
|
|
+
|
|
|
+输出关键词列表，按重要性排序（最核心的在前）。
|
|
|
+""".strip()
|
|
|
+
|
|
|
+class KeywordList(BaseModel):
|
|
|
+ """关键词列表"""
|
|
|
+ keywords: list[str] = Field(..., description="提取的关键词，按重要性排序")
|
|
|
+ reasoning: str = Field(..., description="提取理由")
|
|
|
+
|
|
|
+keyword_extractor = Agent[None](
|
|
|
+ name="关键词提取专家",
|
|
|
+ instructions=keyword_extraction_instructions,
|
|
|
+ model=get_model(MODEL_NAME),
|
|
|
+ output_type=KeywordList,
|
|
|
+)
|
|
|
+
|
|
|
+
|
|
|
+# ============================================================================
|
|
|
+# Agent 2: 层级探索分析专家
|
|
|
+# ============================================================================
|
|
|
+level_analysis_instructions = """
|
|
|
+你是搜索空间探索分析专家。基于当前层级的探索结果，决定下一步行动。
|
|
|
+
|
|
|
+## 你的任务
|
|
|
+
|
|
|
+分析当前已探索的词汇空间，判断：
|
|
|
+1. **发现了什么有价值的信号？**
|
|
|
+2. **是否已经可以评估候选了？**
|
|
|
+3. **如果还不够，下一层应该探索什么组合？**
|
|
|
+
|
|
|
+## 分析维度
|
|
|
+
|
|
|
+### 1. 信号识别（最重要）
|
|
|
+
|
|
|
+看推荐词里**出现了什么主题**：
|
|
|
+
|
|
|
+**关键问题：**
|
|
|
+- 哪些推荐词**最接近原始需求**？
|
|
|
+- 哪些推荐词**揭示了有价值的方向**（即使不完全匹配）？
|
|
|
+- 哪些推荐词可以作为**下一层探索的桥梁**？
|
|
|
+- 系统对哪些概念理解得好？哪些理解偏了？
|
|
|
+
|
|
|
+### 2. 组合策略
|
|
|
+
|
|
|
+基于发现的信号，设计下一层探索：
|
|
|
+
|
|
|
+**组合类型：**
|
|
|
+
|
|
|
+a) **关键词直接组合**
|
|
|
+ - 两个关键词组合成新query
|
|
|
+
|
|
|
+b) **利用推荐词作为桥梁**（重要！）
|
|
|
+ - 发现某个推荐词很有价值 → 直接探索这个推荐词
|
|
|
+ - 或在推荐词基础上加其他关键词
|
|
|
+
|
|
|
+c) **跨层级组合**
|
|
|
+ - 结合多层发现的有价值推荐词
|
|
|
+ - 组合成更复杂的query
|
|
|
+
|
|
|
+### 3. 停止条件
|
|
|
+
|
|
|
+**何时可以评估候选？**
|
|
|
+
|
|
|
+满足以下之一：
|
|
|
+- 推荐词中出现了**明确包含原始需求多个核心要素的query**
|
|
|
+- 已经探索到**足够复杂的组合**（3-4个关键词），且推荐词相关
|
|
|
+- 探索了**3-4层**，信息已经足够丰富
|
|
|
+
|
|
|
+**何时继续探索？**
|
|
|
+- 当前推荐词太泛，没有接近原始需求
|
|
|
+- 发现了有价值的信号，但需要进一步组合验证
|
|
|
+- 层数还少（< 3层）
|
|
|
+
|
|
|
+## 输出要求
|
|
|
+
|
|
|
+### 1. key_findings
|
|
|
+总结当前层发现的关键信息，包括：
|
|
|
+- 哪些推荐词最有价值？
|
|
|
+- 系统对哪些概念理解得好/不好？
|
|
|
+- 发现了什么意外的方向？
|
|
|
+
|
|
|
+### 2. promising_signals
|
|
|
+列出最有价值的推荐词（来自任何已探索的query），每个说明为什么有价值
|
|
|
+
|
|
|
+### 3. should_evaluate_now
|
|
|
+是否已经可以开始评估候选了？true/false
|
|
|
+
|
|
|
+### 4. candidates_to_evaluate
|
|
|
+如果should_evaluate_now=true，列出应该评估的候选query
|
|
|
+- 可以是推荐词
|
|
|
+- 可以是自己构造的组合
|
|
|
+
|
|
|
+### 5. next_combinations
|
|
|
+如果should_evaluate_now=false，列出下一层应该探索的query组合
|
|
|
+
|
|
|
+### 6. reasoning
|
|
|
+详细的推理过程
|
|
|
+
|
|
|
+## 重要原则
|
|
|
+
|
|
|
+1. **不要过早评估**：至少探索2层，除非第一层就发现了完美匹配
|
|
|
+2. **充分利用推荐词**：推荐词是系统给的提示，要善用
|
|
|
+3. **保持探索方向的多样性**：不要只盯着一个方向
|
|
|
+4. **识别死胡同**：如果某个方向的推荐词一直不相关，果断放弃
|
|
|
+""".strip()
|
|
|
+
|
|
|
+class PromisingSignal(BaseModel):
|
|
|
+ """有价值的推荐词信号"""
|
|
|
+ query: str = Field(..., description="推荐词")
|
|
|
+ from_level: int = Field(..., description="来自哪一层")
|
|
|
+ reason: str = Field(..., description="为什么有价值")
|
|
|
+
|
|
|
+class LevelAnalysis(BaseModel):
|
|
|
+ """层级分析结果"""
|
|
|
+ key_findings: str = Field(..., description="当前层的关键发现")
|
|
|
+ promising_signals: list[PromisingSignal] = Field(..., description="有价值的推荐词信号")
|
|
|
+ should_evaluate_now: bool = Field(..., description="是否应该开始评估候选")
|
|
|
+ candidates_to_evaluate: list[str] = Field(default_factory=list, description="如果should_evaluate_now=true，要评估的候选query列表")
|
|
|
+ next_combinations: list[str] = Field(default_factory=list, description="如果should_evaluate_now=false，下一层要探索的query组合")
|
|
|
+ reasoning: str = Field(..., description="详细的推理过程")
|
|
|
+
|
|
|
+level_analyzer = Agent[None](
|
|
|
+ name="层级探索分析专家",
|
|
|
+ instructions=level_analysis_instructions,
|
|
|
+ model=get_model(MODEL_NAME),
|
|
|
+ output_type=LevelAnalysis,
|
|
|
+)
|
|
|
+
|
|
|
+
|
|
|
+# ============================================================================
|
|
|
+# Agent 3: 评估专家（简化版：意图匹配 + 相关性评分）
|
|
|
+# ============================================================================
|
|
|
+eval_instructions = """
|
|
|
+你是搜索query评估专家。给定原始问题和推荐query，评估两个维度。
|
|
|
+
|
|
|
+## 评估目标
|
|
|
+
|
|
|
+用这个推荐query搜索，能否找到满足原始需求的内容？
|
|
|
+
|
|
|
+## 两层评分
|
|
|
+
|
|
|
+### 1. intent_match（意图匹配）= true/false
|
|
|
+
|
|
|
+推荐query的**使用意图**是否与原问题一致？
|
|
|
+
|
|
|
+**核心问题：用户搜索这个推荐词，想做什么？**
|
|
|
+
|
|
|
+**判断标准：**
|
|
|
+- 原问题意图：找方法？找教程？找资源/素材？找工具？看作品？
|
|
|
+- 推荐词意图：如果用户搜索这个词，他的目的是什么？
|
|
|
+
|
|
|
+**示例：**
|
|
|
+- 原问题意图="找素材"
|
|
|
+ - ✅ true: "素材下载"、"素材网站"、"免费素材"（都是获取素材）
|
|
|
+ - ❌ false: "素材制作教程"、"如何制作素材"（意图变成学习了）
|
|
|
+
|
|
|
+- 原问题意图="学教程"
|
|
|
+ - ✅ true: "教程视频"、"教学步骤"、"入门指南"
|
|
|
+ - ❌ false: "成品展示"、"作品欣赏"（意图变成看作品了）
|
|
|
+
|
|
|
+**评分：**
|
|
|
+- true = 意图一致，搜索推荐词能达到原问题的目的
|
|
|
+- false = 意图改变，搜索推荐词无法达到原问题的目的
|
|
|
+
|
|
|
+### 2. relevance_score（相关性）= 0-1 连续分数
|
|
|
+
|
|
|
+推荐query在**主题、要素、属性**上与原问题的相关程度？
|
|
|
+
|
|
|
+**评估维度：**
|
|
|
+- 主题相关：核心主题是否匹配？（如：摄影、旅游、美食）
|
|
|
+- 要素覆盖：关键要素保留了多少？（如：地域、时间、对象、工具）
|
|
|
+- 属性匹配：质量、风格、特色等属性是否保留？
|
|
|
+
|
|
|
+**评分参考：**
|
|
|
+- 0.9-1.0 = 几乎完美匹配，所有核心要素都保留
|
|
|
+- 0.7-0.8 = 高度相关，核心要素保留，少数次要要素缺失
|
|
|
+- 0.5-0.6 = 中度相关，主题匹配但多个要素缺失
|
|
|
+- 0.3-0.4 = 低度相关，只有部分主题相关
|
|
|
+- 0-0.2 = 基本不相关
|
|
|
+
|
|
|
+## 评估策略
|
|
|
+
|
|
|
+1. **先判断 intent_match**：意图不匹配直接 false，无论相关性多高
|
|
|
+2. **再评估 relevance_score**：在意图匹配的前提下，计算相关性
|
|
|
+
|
|
|
+## 输出要求
|
|
|
+
|
|
|
+- intent_match: true/false
|
|
|
+- relevance_score: 0-1 的浮点数
|
|
|
+- reason: 详细的评估理由，需要说明：
|
|
|
+ - 原问题的意图是什么
|
|
|
+ - 推荐词的意图是什么
|
|
|
+ - 为什么判断意图匹配/不匹配
|
|
|
+ - 相关性分数的依据（哪些要素保留/缺失）
|
|
|
+""".strip()
|
|
|
+
|
|
|
+class RelevanceEvaluation(BaseModel):
|
|
|
+ """评估反馈模型 - 意图匹配 + 相关性"""
|
|
|
+ intent_match: bool = Field(..., description="意图是否匹配")
|
|
|
+ relevance_score: float = Field(..., description="相关性分数 0-1，分数越高越相关")
|
|
|
+ reason: str = Field(..., description="评估理由，需说明意图判断和相关性依据")
|
|
|
+
|
|
|
+evaluator = Agent[None](
|
|
|
+ name="评估专家",
|
|
|
+ instructions=eval_instructions,
|
|
|
+ model=get_model(MODEL_NAME),
|
|
|
+ output_type=RelevanceEvaluation,
|
|
|
+)
|
|
|
+
|
|
|
+
|
|
|
+# ============================================================================
|
|
|
+# 日志辅助函数
|
|
|
+# ============================================================================
|
|
|
+
|
|
|
+def add_step(context: RunContext, step_name: str, step_type: str, data: dict):
|
|
|
+ """添加步骤记录"""
|
|
|
+ step = {
|
|
|
+ "step_number": len(context.steps) + 1,
|
|
|
+ "step_name": step_name,
|
|
|
+ "step_type": step_type,
|
|
|
+ "timestamp": datetime.now().isoformat(),
|
|
|
+ "data": data
|
|
|
+ }
|
|
|
+ context.steps.append(step)
|
|
|
+ return step
|
|
|
+
|
|
|
+
|
|
|
+# ============================================================================
|
|
|
+# 核心函数
|
|
|
+# ============================================================================
|
|
|
+
|
|
|
+async def extract_keywords(q: str, context: RunContext) -> KeywordList:
|
|
|
+ """提取关键词"""
|
|
|
+ print("\n[步骤 1] 正在提取关键词...")
|
|
|
+ result = await Runner.run(keyword_extractor, q)
|
|
|
+ keyword_list: KeywordList = result.final_output
|
|
|
+ print(f"提取的关键词：{keyword_list.keywords}")
|
|
|
+ print(f"提取理由：{keyword_list.reasoning}")
|
|
|
+
|
|
|
+ # 记录步骤
|
|
|
+ add_step(context, "提取关键词", "keyword_extraction", {
|
|
|
+ "input_question": q,
|
|
|
+ "keywords": keyword_list.keywords,
|
|
|
+ "reasoning": keyword_list.reasoning
|
|
|
+ })
|
|
|
+
|
|
|
+ return keyword_list
|
|
|
+
|
|
|
+
|
|
|
+async def explore_level(queries: list[str], level_num: int, context: RunContext) -> dict:
|
|
|
+ """探索一个层级（并发获取所有query的推荐词）"""
|
|
|
+ step_num = len(context.steps) + 1
|
|
|
+ print(f"\n{'='*60}")
|
|
|
+ print(f"[步骤 {step_num}] Level {level_num} 探索：{len(queries)} 个query")
|
|
|
+ print(f"{'='*60}")
|
|
|
+
|
|
|
+ xiaohongshu_api = XiaohongshuSearchRecommendations()
|
|
|
+
|
|
|
+ # 并发获取所有推荐词
|
|
|
+ async def get_single_sug(query: str):
|
|
|
+ print(f" 探索: {query}")
|
|
|
+ suggestions = xiaohongshu_api.get_recommendations(keyword=query)
|
|
|
+ print(f" → {len(suggestions) if suggestions else 0} 个推荐词")
|
|
|
+ return {
|
|
|
+ "query": query,
|
|
|
+ "suggestions": suggestions or []
|
|
|
+ }
|
|
|
+
|
|
|
+ results = await asyncio.gather(*[get_single_sug(q) for q in queries])
|
|
|
+
|
|
|
+ level_data = {
|
|
|
+ "level": level_num,
|
|
|
+ "timestamp": datetime.now().isoformat(),
|
|
|
+ "queries": results
|
|
|
+ }
|
|
|
+
|
|
|
+ context.exploration_levels.append(level_data)
|
|
|
+
|
|
|
+ # 记录步骤
|
|
|
+ add_step(context, f"Level {level_num} 探索", "level_exploration", {
|
|
|
+ "level": level_num,
|
|
|
+ "input_queries": queries,
|
|
|
+ "query_count": len(queries),
|
|
|
+ "results": results,
|
|
|
+ "total_suggestions": sum(len(r['suggestions']) for r in results)
|
|
|
+ })
|
|
|
+
|
|
|
+ return level_data
|
|
|
+
|
|
|
+
|
|
|
+async def analyze_level(level_data: dict, all_levels: list[dict], original_question: str, context: RunContext) -> LevelAnalysis:
|
|
|
+ """分析当前层级，决定下一步"""
|
|
|
+ step_num = len(context.steps) + 1
|
|
|
+ print(f"\n[步骤 {step_num}] 正在分析 Level {level_data['level']}...")
|
|
|
+
|
|
|
+ # 构造输入
|
|
|
+ analysis_input = f"""
|
|
|
+<原始问题>
|
|
|
+{original_question}
|
|
|
+</原始问题>
|
|
|
+
|
|
|
+<已探索的所有层级>
|
|
|
+{json.dumps(all_levels, ensure_ascii=False, indent=2)}
|
|
|
+</已探索的所有层级>
|
|
|
+
|
|
|
+<当前层级>
|
|
|
+Level {level_data['level']}
|
|
|
+{json.dumps(level_data['queries'], ensure_ascii=False, indent=2)}
|
|
|
+</当前层级>
|
|
|
+
|
|
|
+请分析当前探索状态，决定下一步行动。
|
|
|
+"""
|
|
|
+
|
|
|
+ result = await Runner.run(level_analyzer, analysis_input)
|
|
|
+ analysis: LevelAnalysis = result.final_output
|
|
|
+
|
|
|
+ print(f"\n分析结果：")
|
|
|
+ print(f" 关键发现：{analysis.key_findings}")
|
|
|
+ print(f" 有价值的信号：{len(analysis.promising_signals)} 个")
|
|
|
+ print(f" 是否评估：{analysis.should_evaluate_now}")
|
|
|
+
|
|
|
+ if analysis.should_evaluate_now:
|
|
|
+ print(f" 候选query：{analysis.candidates_to_evaluate}")
|
|
|
+ else:
|
|
|
+ print(f" 下一层探索：{analysis.next_combinations}")
|
|
|
+
|
|
|
+ # 保存分析结果
|
|
|
+ context.level_analyses.append({
|
|
|
+ "level": level_data['level'],
|
|
|
+ "timestamp": datetime.now().isoformat(),
|
|
|
+ "analysis": analysis.model_dump()
|
|
|
+ })
|
|
|
+
|
|
|
+ # 记录步骤
|
|
|
+ add_step(context, f"Level {level_data['level']} 分析", "level_analysis", {
|
|
|
+ "level": level_data['level'],
|
|
|
+ "key_findings": analysis.key_findings,
|
|
|
+ "promising_signals_count": len(analysis.promising_signals),
|
|
|
+ "promising_signals": [s.model_dump() for s in analysis.promising_signals],
|
|
|
+ "should_evaluate_now": analysis.should_evaluate_now,
|
|
|
+ "candidates_to_evaluate": analysis.candidates_to_evaluate if analysis.should_evaluate_now else [],
|
|
|
+ "next_combinations": analysis.next_combinations if not analysis.should_evaluate_now else [],
|
|
|
+ "reasoning": analysis.reasoning
|
|
|
+ })
|
|
|
+
|
|
|
+ return analysis
|
|
|
+
|
|
|
+
|
|
|
+async def evaluate_candidates(candidates: list[str], original_question: str, context: RunContext) -> list[dict]:
|
|
|
+ """评估候选query"""
|
|
|
+ step_num = len(context.steps) + 1
|
|
|
+ print(f"\n{'='*60}")
|
|
|
+ print(f"[步骤 {step_num}] 评估 {len(candidates)} 个候选query")
|
|
|
+ print(f"{'='*60}")
|
|
|
+
|
|
|
+ xiaohongshu_api = XiaohongshuSearchRecommendations()
|
|
|
+
|
|
|
+ async def evaluate_single_candidate(candidate: str):
|
|
|
+ print(f"\n评估候选：{candidate}")
|
|
|
+
|
|
|
+ # 1. 获取推荐词
|
|
|
+ suggestions = xiaohongshu_api.get_recommendations(keyword=candidate)
|
|
|
+ print(f" 获取到 {len(suggestions) if suggestions else 0} 个推荐词")
|
|
|
+
|
|
|
+ if not suggestions:
|
|
|
+ return {
|
|
|
+ "candidate": candidate,
|
|
|
+ "suggestions": [],
|
|
|
+ "evaluations": []
|
|
|
+ }
|
|
|
+
|
|
|
+ # 2. 评估每个推荐词
|
|
|
+ async def eval_single_sug(sug: str):
|
|
|
+ eval_input = f"""
|
|
|
+<原始问题>
|
|
|
+{original_question}
|
|
|
+</原始问题>
|
|
|
+
|
|
|
+<待评估的推荐query>
|
|
|
+{sug}
|
|
|
+</待评估的推荐query>
|
|
|
+
|
|
|
+请评估该推荐query：
|
|
|
+1. intent_match: 意图是否匹配（true/false）
|
|
|
+2. relevance_score: 相关性分数（0-1）
|
|
|
+3. reason: 详细的评估理由
|
|
|
+"""
|
|
|
+ result = await Runner.run(evaluator, eval_input)
|
|
|
+ evaluation: RelevanceEvaluation = result.final_output
|
|
|
+ return {
|
|
|
+ "query": sug,
|
|
|
+ "intent_match": evaluation.intent_match,
|
|
|
+ "relevance_score": evaluation.relevance_score,
|
|
|
+ "reason": evaluation.reason,
|
|
|
+ }
|
|
|
+
|
|
|
+ evaluations = await asyncio.gather(*[eval_single_sug(s) for s in suggestions])
|
|
|
+
|
|
|
+ return {
|
|
|
+ "candidate": candidate,
|
|
|
+ "suggestions": suggestions,
|
|
|
+ "evaluations": evaluations
|
|
|
+ }
|
|
|
+
|
|
|
+ results = await asyncio.gather(*[evaluate_single_candidate(c) for c in candidates])
|
|
|
+
|
|
|
+ context.evaluation_results = results
|
|
|
+
|
|
|
+ # 记录步骤
|
|
|
+ add_step(context, "评估候选query", "candidate_evaluation", {
|
|
|
+ "candidate_count": len(candidates),
|
|
|
+ "candidates": candidates,
|
|
|
+ "results": results,
|
|
|
+ "total_evaluations": sum(len(r['evaluations']) for r in results)
|
|
|
+ })
|
|
|
+
|
|
|
+ return results
|
|
|
+
|
|
|
+
|
|
|
+def find_qualified_queries(evaluation_results: list[dict], min_relevance_score: float = 0.7) -> list[dict]:
|
|
|
+ """
|
|
|
+ 查找所有合格的query
|
|
|
+
|
|
|
+ 筛选标准：
|
|
|
+ 1. intent_match = True（必须满足）
|
|
|
+ 2. relevance_score >= min_relevance_score
|
|
|
+
|
|
|
+ 返回：按 relevance_score 降序排列
|
|
|
+ """
|
|
|
+ all_qualified = []
|
|
|
+
|
|
|
+ for result in evaluation_results:
|
|
|
+ for eval_item in result.get("evaluations", []):
|
|
|
+ if (eval_item['intent_match'] is True
|
|
|
+ and eval_item['relevance_score'] >= min_relevance_score):
|
|
|
+ all_qualified.append({
|
|
|
+ "from_candidate": result["candidate"],
|
|
|
+ **eval_item
|
|
|
+ })
|
|
|
+
|
|
|
+ # 按relevance_score降序排列
|
|
|
+ return sorted(all_qualified, key=lambda x: x['relevance_score'], reverse=True)
|
|
|
+
|
|
|
+
|
|
|
+# ============================================================================
|
|
|
+# 主流程
|
|
|
+# ============================================================================
|
|
|
+
|
|
|
+async def progressive_exploration(context: RunContext, max_levels: int = 4) -> dict:
|
|
|
+ """
|
|
|
+ 渐进式广度探索流程
|
|
|
+
|
|
|
+ Args:
|
|
|
+ context: 运行上下文
|
|
|
+ max_levels: 最大探索层数，默认4
|
|
|
+
|
|
|
+ 返回格式：
|
|
|
+ {
|
|
|
+ "success": True/False,
|
|
|
+ "results": [...],
|
|
|
+ "message": "..."
|
|
|
+ }
|
|
|
+ """
|
|
|
+
|
|
|
+ # 阶段1：提取关键词（从原始问题提取）
|
|
|
+ keyword_result = await extract_keywords(context.q, context)
|
|
|
+ context.keywords = keyword_result.keywords
|
|
|
+
|
|
|
+ # 阶段2：渐进式探索
|
|
|
+ current_level = 1
|
|
|
+
|
|
|
+ # Level 1：单个关键词
|
|
|
+ level_1_queries = context.keywords[:7] # 限制最多7个关键词
|
|
|
+ level_1_data = await explore_level(level_1_queries, current_level, context)
|
|
|
+
|
|
|
+ # 分析Level 1
|
|
|
+ analysis_1 = await analyze_level(level_1_data, context.exploration_levels, context.q, context)
|
|
|
+
|
|
|
+ if analysis_1.should_evaluate_now:
|
|
|
+ # 直接评估
|
|
|
+ eval_results = await evaluate_candidates(analysis_1.candidates_to_evaluate, context.q, context)
|
|
|
+ qualified = find_qualified_queries(eval_results, min_relevance_score=0.7)
|
|
|
+
|
|
|
+ if qualified:
|
|
|
+ return {
|
|
|
+ "success": True,
|
|
|
+ "results": qualified,
|
|
|
+ "message": f"Level 1 即找到 {len(qualified)} 个合格query"
|
|
|
+ }
|
|
|
+
|
|
|
+ # Level 2 及以后：迭代探索
|
|
|
+ for level_num in range(2, max_levels + 1):
|
|
|
+ # 获取上一层的分析结果
|
|
|
+ prev_analysis: LevelAnalysis = context.level_analyses[-1]["analysis"]
|
|
|
+ prev_analysis = LevelAnalysis(**prev_analysis) # 转回对象
|
|
|
+
|
|
|
+ if not prev_analysis.next_combinations:
|
|
|
+ print(f"\nLevel {level_num-1} 分析后无需继续探索")
|
|
|
+ break
|
|
|
+
|
|
|
+ # 探索当前层
|
|
|
+ level_data = await explore_level(prev_analysis.next_combinations, level_num, context)
|
|
|
+
|
|
|
+ # 分析当前层
|
|
|
+ analysis = await analyze_level(level_data, context.exploration_levels, context.q, context)
|
|
|
+
|
|
|
+ if analysis.should_evaluate_now:
|
|
|
+ # 评估候选
|
|
|
+ eval_results = await evaluate_candidates(analysis.candidates_to_evaluate, context.q, context)
|
|
|
+ qualified = find_qualified_queries(eval_results, min_relevance_score=0.7)
|
|
|
+
|
|
|
+ if qualified:
|
|
|
+ return {
|
|
|
+ "success": True,
|
|
|
+ "results": qualified,
|
|
|
+ "message": f"Level {level_num} 找到 {len(qualified)} 个合格query"
|
|
|
+ }
|
|
|
+
|
|
|
+ # 所有层探索完，降低标准
|
|
|
+ print(f"\n{'='*60}")
|
|
|
+ print(f"探索完 {max_levels} 层，降低标准（relevance_score >= 0.5）")
|
|
|
+ print(f"{'='*60}")
|
|
|
+
|
|
|
+ if context.evaluation_results:
|
|
|
+ acceptable = find_qualified_queries(context.evaluation_results, min_relevance_score=0.5)
|
|
|
+ if acceptable:
|
|
|
+ return {
|
|
|
+ "success": True,
|
|
|
+ "results": acceptable,
|
|
|
+ "message": f"找到 {len(acceptable)} 个可接受query（soft_score >= 0.5）"
|
|
|
+ }
|
|
|
+
|
|
|
+ # 完全失败
|
|
|
+ return {
|
|
|
+ "success": False,
|
|
|
+ "results": [],
|
|
|
+ "message": "探索完所有层级，未找到合格的推荐词"
|
|
|
+ }
|
|
|
+
|
|
|
+
|
|
|
+# ============================================================================
|
|
|
+# 输出格式化
|
|
|
+# ============================================================================
|
|
|
+
|
|
|
+def format_output(optimization_result: dict, context: RunContext) -> str:
|
|
|
+ """格式化输出结果"""
|
|
|
+ results = optimization_result.get("results", [])
|
|
|
+
|
|
|
+ output = f"原始问题：{context.q}\n"
|
|
|
+ output += f"提取的关键词：{', '.join(context.keywords or [])}\n"
|
|
|
+ output += f"探索层数：{len(context.exploration_levels)}\n"
|
|
|
+ output += f"状态：{optimization_result['message']}\n\n"
|
|
|
+
|
|
|
+ if optimization_result["success"] and results:
|
|
|
+ output += "合格的推荐query（按relevance_score降序）：\n"
|
|
|
+ for i, result in enumerate(results, 1):
|
|
|
+ output += f"\n{i}. {result['query']}\n"
|
|
|
+ output += f" - 来自候选：{result['from_candidate']}\n"
|
|
|
+ output += f" - 意图匹配：{result['intent_match']} (True=意图一致)\n"
|
|
|
+ output += f" - 相关性分数：{result['relevance_score']:.2f} (0-1，越高越相关)\n"
|
|
|
+ output += f" - 评估理由：{result['reason']}\n"
|
|
|
+ else:
|
|
|
+ output += "结果：未找到合格推荐query\n"
|
|
|
+ if context.level_analyses:
|
|
|
+ last_analysis = context.level_analyses[-1]["analysis"]
|
|
|
+ output += f"\n最后一层分析：\n{last_analysis.get('key_findings', 'N/A')}\n"
|
|
|
+
|
|
|
+ return output.strip()
|
|
|
+
|
|
|
+
|
|
|
+# ============================================================================
|
|
|
+# 主函数
|
|
|
+# ============================================================================
|
|
|
+
|
|
|
+async def main(input_dir: str, max_levels: int = 4):
|
|
|
+ current_time, log_url = set_trace()
|
|
|
+
|
|
|
+ # 从目录中读取固定文件名
|
|
|
+ input_context_file = os.path.join(input_dir, 'context.md')
|
|
|
+ input_q_file = os.path.join(input_dir, 'q.md')
|
|
|
+
|
|
|
+ q_context = read_file_as_string(input_context_file)
|
|
|
+ q = read_file_as_string(input_q_file)
|
|
|
+ q_with_context = f"""
|
|
|
+<需求上下文>
|
|
|
+{q_context}
|
|
|
+</需求上下文>
|
|
|
+<当前问题>
|
|
|
+{q}
|
|
|
+</当前问题>
|
|
|
+""".strip()
|
|
|
+
|
|
|
+ # 获取当前文件名作为版本
|
|
|
+ version = os.path.basename(__file__)
|
|
|
+ version_name = os.path.splitext(version)[0]
|
|
|
+
|
|
|
+ # 日志保存目录
|
|
|
+ log_dir = os.path.join(input_dir, "output", version_name, current_time)
|
|
|
+
|
|
|
+ run_context = RunContext(
|
|
|
+ version=version,
|
|
|
+ input_files={
|
|
|
+ "input_dir": input_dir,
|
|
|
+ "context_file": input_context_file,
|
|
|
+ "q_file": input_q_file,
|
|
|
+ },
|
|
|
+ q_with_context=q_with_context,
|
|
|
+ q_context=q_context,
|
|
|
+ q=q,
|
|
|
+ log_dir=log_dir,
|
|
|
+ log_url=log_url,
|
|
|
+ )
|
|
|
+
|
|
|
+ # 执行渐进式探索
|
|
|
+ optimization_result = await progressive_exploration(run_context, max_levels=max_levels)
|
|
|
+
|
|
|
+ # 格式化输出
|
|
|
+ final_output = format_output(optimization_result, run_context)
|
|
|
+ print(f"\n{'='*60}")
|
|
|
+ print("最终结果")
|
|
|
+ print(f"{'='*60}")
|
|
|
+ print(final_output)
|
|
|
+
|
|
|
+ # 保存结果
|
|
|
+ run_context.optimization_result = optimization_result
|
|
|
+ run_context.final_output = final_output
|
|
|
+
|
|
|
+ # 记录最终输出步骤（保存完整的结果详情）
|
|
|
+ qualified_results = optimization_result.get("results", [])
|
|
|
+ add_step(run_context, "生成最终结果", "final_result", {
|
|
|
+ "success": optimization_result["success"],
|
|
|
+ "message": optimization_result["message"],
|
|
|
+ "qualified_query_count": len(qualified_results),
|
|
|
+ "qualified_queries": [r["query"] for r in qualified_results], # 保存所有合格query
|
|
|
+ "qualified_results_detail": [ # 保存完整的评估详情
|
|
|
+ {
|
|
|
+ "rank": idx + 1,
|
|
|
+ "query": r["query"],
|
|
|
+ "from_candidate": r["from_candidate"],
|
|
|
+ "intent_match": r["intent_match"],
|
|
|
+ "relevance_score": r["relevance_score"],
|
|
|
+ "reason": r["reason"]
|
|
|
+ }
|
|
|
+ for idx, r in enumerate(qualified_results)
|
|
|
+ ],
|
|
|
+ "final_output": final_output
|
|
|
+ })
|
|
|
+
|
|
|
+ # 保存 RunContext 到 log_dir
|
|
|
+ os.makedirs(run_context.log_dir, exist_ok=True)
|
|
|
+ context_file_path = os.path.join(run_context.log_dir, "run_context.json")
|
|
|
+ with open(context_file_path, "w", encoding="utf-8") as f:
|
|
|
+ json.dump(run_context.model_dump(), f, ensure_ascii=False, indent=2)
|
|
|
+ print(f"\nRunContext saved to: {context_file_path}")
|
|
|
+
|
|
|
+ # 保存步骤化日志（更直观的格式）
|
|
|
+ steps_file_path = os.path.join(run_context.log_dir, "steps.json")
|
|
|
+ with open(steps_file_path, "w", encoding="utf-8") as f:
|
|
|
+ json.dump(run_context.steps, f, ensure_ascii=False, indent=2)
|
|
|
+ print(f"Steps log saved to: {steps_file_path}")
|
|
|
+
|
|
|
+ # 生成步骤化的可读文本日志
|
|
|
+ steps_text_path = os.path.join(run_context.log_dir, "steps.md")
|
|
|
+ with open(steps_text_path, "w", encoding="utf-8") as f:
|
|
|
+ f.write(f"# 执行步骤日志\n\n")
|
|
|
+ f.write(f"**原始问题**: {run_context.q}\n\n")
|
|
|
+ f.write(f"**执行版本**: {run_context.version}\n\n")
|
|
|
+ f.write(f"**总步骤数**: {len(run_context.steps)}\n\n")
|
|
|
+ f.write("---\n\n")
|
|
|
+
|
|
|
+ for step in run_context.steps:
|
|
|
+ f.write(f"## 步骤 {step['step_number']}: {step['step_name']}\n\n")
|
|
|
+ f.write(f"**类型**: `{step['step_type']}`\n\n")
|
|
|
+ f.write(f"**时间**: {step['timestamp']}\n\n")
|
|
|
+
|
|
|
+ # 根据不同类型格式化数据
|
|
|
+ if step['step_type'] == 'keyword_extraction':
|
|
|
+ f.write(f"**提取的关键词**: {', '.join(step['data']['keywords'])}\n\n")
|
|
|
+ f.write(f"**提取理由**: {step['data']['reasoning']}\n\n")
|
|
|
+
|
|
|
+ elif step['step_type'] == 'level_exploration':
|
|
|
+ f.write(f"**探索层级**: Level {step['data']['level']}\n\n")
|
|
|
+ f.write(f"**输入query数量**: {step['data']['query_count']}\n\n")
|
|
|
+ f.write(f"**总推荐词数**: {step['data']['total_suggestions']}\n\n")
|
|
|
+ f.write(f"**探索的query**: {', '.join(step['data']['input_queries'])}\n\n")
|
|
|
+
|
|
|
+ elif step['step_type'] == 'level_analysis':
|
|
|
+ f.write(f"**关键发现**: {step['data']['key_findings']}\n\n")
|
|
|
+ f.write(f"**有价值信号数**: {step['data']['promising_signals_count']}\n\n")
|
|
|
+ f.write(f"**是否评估**: {step['data']['should_evaluate_now']}\n\n")
|
|
|
+ if step['data']['should_evaluate_now']:
|
|
|
+ f.write(f"**候选query**: {', '.join(step['data']['candidates_to_evaluate'])}\n\n")
|
|
|
+ else:
|
|
|
+ f.write(f"**下一层探索**: {', '.join(step['data']['next_combinations'])}\n\n")
|
|
|
+
|
|
|
+ elif step['step_type'] == 'candidate_evaluation':
|
|
|
+ f.write(f"**评估候选数**: {step['data']['candidate_count']}\n\n")
|
|
|
+ f.write(f"**候选query**: {', '.join(step['data']['candidates'])}\n\n")
|
|
|
+ f.write(f"**总评估数**: {step['data']['total_evaluations']}\n\n")
|
|
|
+
|
|
|
+ elif step['step_type'] == 'final_result':
|
|
|
+ f.write(f"**执行状态**: {'✅ 成功' if step['data']['success'] else '❌ 失败'}\n\n")
|
|
|
+ f.write(f"**结果消息**: {step['data']['message']}\n\n")
|
|
|
+ f.write(f"**合格query数量**: {step['data']['qualified_query_count']}\n\n")
|
|
|
+
|
|
|
+ # 显示详细的评估结果
|
|
|
+ if step['data'].get('qualified_results_detail'):
|
|
|
+ f.write(f"### 合格的query详情\n\n")
|
|
|
+ for result in step['data']['qualified_results_detail']:
|
|
|
+ f.write(f"#### {result['rank']}. {result['query']}\n\n")
|
|
|
+ f.write(f"- **来自候选**: {result['from_candidate']}\n")
|
|
|
+ f.write(f"- **意图匹配**: {'✅ 是' if result['intent_match'] else '❌ 否'}\n")
|
|
|
+ f.write(f"- **相关性分数**: {result['relevance_score']:.2f}\n")
|
|
|
+ f.write(f"- **评估理由**: {result['reason']}\n\n")
|
|
|
+ elif step['data']['qualified_queries']:
|
|
|
+ # 兼容旧格式（如果没有详情）
|
|
|
+ f.write(f"**合格的query列表**:\n")
|
|
|
+ for idx, q in enumerate(step['data']['qualified_queries'], 1):
|
|
|
+ f.write(f" {idx}. {q}\n")
|
|
|
+ f.write("\n")
|
|
|
+
|
|
|
+ f.write(f"### 完整输出\n\n```\n{step['data']['final_output']}\n```\n\n")
|
|
|
+
|
|
|
+ f.write("---\n\n")
|
|
|
+
|
|
|
+ print(f"Steps markdown saved to: {steps_text_path}")
|
|
|
+
|
|
|
+
|
|
|
+if __name__ == "__main__":
|
|
|
+ parser = argparse.ArgumentParser(description="搜索query优化工具 - v6.1 意图匹配+相关性评分版")
|
|
|
+ parser.add_argument(
|
|
|
+ "--input-dir",
|
|
|
+ type=str,
|
|
|
+ default="input/简单扣图",
|
|
|
+ help="输入目录路径，默认: input/简单扣图"
|
|
|
+ )
|
|
|
+ parser.add_argument(
|
|
|
+ "--max-levels",
|
|
|
+ type=int,
|
|
|
+ default=4,
|
|
|
+ help="最大探索层数，默认: 4"
|
|
|
+ )
|
|
|
+ args = parser.parse_args()
|
|
|
+
|
|
|
+ asyncio.run(main(args.input_dir, max_levels=args.max_levels))
|
