add

sug_v1_2.py

@@ -0,0 +1,179 @@
+import asyncio
+
+from agents import Agent, Runner, function_tool
+from lib.my_trace import set_trace
+from lib.utils import read_file_as_string
+from script.search_recommendations.xiaohongshu_search_recommendations import XiaohongshuSearchRecommendations
+
+@function_tool
+def get_query_suggestions(query: str):
+    """Fetch search recommendations from Xiaohongshu."""
+    xiaohongshu_api = XiaohongshuSearchRecommendations()
+    query_suggestions = xiaohongshu_api.get_recommendations(keyword=query)['result']['data']['data']
+    return query_suggestions
+
+@function_tool
+def modify_query(original_query: str, operation_type: str, new_query: str, reason: str):
+    """
+    Modify the search query with a specific operation.
+
+    Args:
+        original_query: The original query before modification
+        operation_type: Type of modification - must be one of: "简化", "扩展", "替换", "组合"
+        new_query: The modified query after applying the operation
+        reason: Detailed explanation of why this modification was made and what insight from previous suggestions led to this change
+
+    Returns:
+        A dict containing the modification record and the new query to use for next search
+    """
+    operation_types = ["简化", "扩展", "替换", "组合"]
+    if operation_type not in operation_types:
+        return {
+            "status": "error",
+            "message": f"Invalid operation_type. Must be one of: {', '.join(operation_types)}"
+        }
+
+    modification_record = {
+        "original_query": original_query,
+        "operation_type": operation_type,
+        "new_query": new_query,
+        "reason": reason,
+    }
+
+    return {
+        "status": "success",
+        "modification_record": modification_record,
+        "new_query": new_query,
+        "message": f"Query modified successfully. Use '{new_query}' for the next search."
+    }
+
+insrtuctions = """
+你是一个专业的搜索query优化专家，擅长通过动态探索找到最符合用户搜索习惯的query。
+
+## 核心任务
+给定原始问题，通过迭代调用搜索推荐接口（get_query_suggestions），找到与原始问题语义等价且更符合平台用户搜索习惯的推荐query。
+
+## 工作流程
+
+### 1. 理解原始问题
+- 仔细阅读<需求上下文>和<当前问题>
+- 提取问题的核心需求和关键概念
+- 明确问题的本质意图（what）、应用场景（where）、实现方式（how）
+
+### 2. 动态探索策略
+采用类似人类搜索的迭代探索方式：
+
+**第一轮尝试：**
+- 使用原始问题直接调用 get_query_suggestions(query="原始问题")
+- 仔细分析返回的推荐词列表
+- 判断是否有与原始问题等价的推荐词
+
+**后续迭代：**
+如果推荐词不满足要求，必须先调用 modify_query 函数记录修改，然后再次搜索：
+
+**工具使用流程：**
+1. 调用 modify_query(original_query, operation_type, new_query, reason)
+2. 使用返回的 new_query 调用 get_query_suggestions
+3. 分析新的推荐词列表
+4. 如果仍不满足，重复步骤1-3
+
+**四种操作类型（operation_type）：**
+- **简化**：删除冗余词汇，提取核心关键词
+- **扩展**
+- **替换**：使用同义词、行业术语或口语化表达
+- **组合**：调整关键词顺序或组合方式
+
+**每次修改的reason必须包含：**
+- 上一轮推荐词给你的启发
+- 为什么这样修改更符合平台用户习惯
+- 与原始问题的关系（确保核心意图不变）
+
+### 3. 等价性判断标准
+推荐词满足以下条件即可视为"与原始问题等价"：
+
+**语义等价：**
+- 能够回答或解决原始问题的核心需求
+- 涵盖原始问题的关键功能或场景
+
+
+### 4. 迭代终止条件
+- **成功终止**：找到至少一个与原始问题等价的推荐query
+- **尝试上限**：最多迭代5轮，避免无限循环
+- **无推荐词**：推荐接口返回空列表或错误
+
+### 5. 输出要求
+成功找到等价query时，输出格式：
+```
+原始问题：[原问题]
+
+优化后的query：[最终找到的等价推荐query]
+
+探索路径：
+1. 第1轮：原始query "[query1]"
+   - 推荐词：[列出关键推荐词]
+   - 判断：不满足，[简要说明原因]
+
+2. 第2轮：modify_query("[query1]", "简化", "[query2]", "[reason]")
+   - 推荐词：[列出关键推荐词]
+   - 判断：不满足，[简要说明原因]
+
+3. 第3轮：modify_query("[query2]", "替换", "[query3]", "[reason]")
+   - 推荐词：[列出关键推荐词]
+   - 判断：满足！找到等价query "[最终query]"
+
+推荐理由：
+- 该query来自平台官方推荐，大概率有结果
+- 与原始问题语义等价：[具体说明]
+- 更符合用户搜索习惯：[具体说明]
+```
+
+未找到等价query时，输出：
+```
+原始问题：[原问题]
+
+探索结果：未找到完全等价的推荐query（已尝试[N]轮）
+
+尝试过的query及修改记录：
+1. "[query1]" (原始)
+2. "[query2]" (简化：[reason])
+3. "[query3]" (替换：[reason])
+...
+
+各轮推荐词分析：
+- 第1轮推荐词偏向：[分析]
+- 第2轮推荐词偏向：[分析]
+...
+
+建议：
+[基于探索结果给出建议，如：
+- 直接使用原问题搜索
+- 使用尝试过的某个query（虽然不完全等价但最接近）
+- 调整搜索策略或平台]
+```
+
+## 注意事项
+- **第一轮必须使用原始问题**：直接调用 get_query_suggestions(query="原始问题")
+- **后续修改必须调用 modify_query**：不能直接用新query调用 get_query_suggestions，必须先通过 modify_query 记录修改
+- **每次调用 get_query_suggestions 后必须仔细分析**：列出关键推荐词，分析它们的特点和偏向
+- **修改要有理有据**：reason参数必须详细说明基于什么推荐词反馈做出此修改
+- **保持核心意图不变**：每次修改都要确认与原始问题的等价性
+- **优先选择简洁、口语化的推荐词**：如果多个推荐词都满足，选择最符合用户习惯的
+""".strip()
+
+agent = Agent(
+    name="Query Optimization Agent",
+    instructions=insrtuctions,
+    tools=[get_query_suggestions, modify_query],
+)
+
+
+async def main():
+    set_trace()
+    user_input = read_file_as_string('input/kg_v1_single.md')
+    result = await Runner.run(agent, input=user_input)
+    print(result.final_output)
+    # The weather in Tokyo is sunny.
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

sug_v2.py

@@ -1,6 +1,7 @@
 import asyncio
 import json
 import os
+import argparse
 from datetime import datetime
 
 from agents import Agent, Runner, function_tool
@@ -28,6 +29,8 @@ class RunContext(BaseModel):
     log_dir: str
     # 中间数据记录 - 按时间顺序记录所有操作
     operations_history: list[dict] = Field(default_factory=list, description="记录所有操作的历史，包括 get_query_suggestions 和 modify_query")
+    # 最终输出结果
+    final_output: str | None = Field(default=None, description="Agent的最终输出结果")
 
 
 eval_insrtuctions = """
@@ -238,12 +241,12 @@ insrtuctions = """
 
 
 
-async def main():
+async def main(input_dir: str):
     current_time, log_url = set_trace()
 
-    # 输入文件路径
-    input_context_file = 'input/kg_v1_single_context.md'
-    input_q_file = 'input/kg_v1_single_q.md'
+    # 从目录中读取固定文件名
+    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)
@@ -255,14 +258,18 @@ async def main():
 {q}
 </当前问题>
 """.strip()
-    log_dir = os.path.join("logs", current_time)
 
     # 获取当前文件名作为版本
     version = os.path.basename(__file__)
+    version_name = os.path.splitext(version)[0]  # 去掉 .py 后缀
+
+    # 日志保存到输入目录的 output/版本/时间戳 目录下
+    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,
         },
@@ -281,6 +288,9 @@ async def main():
     result = await Runner.run(agent, input=q_with_context,  context = run_context,)
     print(result.final_output)
 
+    # 保存最终输出到 RunContext
+    run_context.final_output = str(result.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")
@@ -290,4 +300,13 @@ async def main():
 
 
 if __name__ == "__main__":
-    asyncio.run(main())
+    parser = argparse.ArgumentParser(description="搜索query优化工具")
+    parser.add_argument(
+        "--input-dir",
+        type=str,
+        default="input/简单扣图",
+        help="输入目录路径，默认: input/简单扣图"
+    )
+    args = parser.parse_args()
+
+    asyncio.run(main(args.input_dir))

sug_v2_1.py

@@ -0,0 +1,311 @@
+import asyncio
+import json
+import os
+from datetime import datetime
+
+from agents import Agent, Runner, function_tool
+from lib.my_trace import set_trace
+from typing import Literal
+from dataclasses import dataclass
+from pydantic import BaseModel, Field
+
+
+from lib.utils import read_file_as_string
+from script.search_recommendations.xiaohongshu_search_recommendations import XiaohongshuSearchRecommendations
+
+from agents import Agent, RunContextWrapper, Runner, function_tool
+
+from pydantic import BaseModel, Field
+
+
+class RunContext(BaseModel):
+    q_with_context: str
+    q_context: str
+    q: str
+    log_url: str
+    log_dir: str
+    # 中间数据记录 - 按时间顺序记录所有操作
+    operations_history: list[dict] = Field(default_factory=list, description="记录所有操作的历史，包括 get_query_suggestions 和 modify_query")
+
+
+eval_insrtuctions = """
+你是一个专业的评估专家，负责评估给定的搜索query是否满足原始问题和需求，给出出评分和简明扼要的理由。
+"""
+
+@dataclass
+class EvaluationFeedback:
+    reason: str=Field(..., description="简明扼要的理由")
+    score: float=Field(..., description="评估结果，1表示等价，0表示不等价，中间值表示部分等价")
+
+evaluator = Agent[None](
+    name="评估专家",
+    instructions=eval_insrtuctions,
+    output_type=EvaluationFeedback,
+)
+
+@function_tool
+async def get_query_suggestions(wrapper: RunContextWrapper[RunContext], query: str):
+    """Fetch search recommendations from Xiaohongshu."""
+    xiaohongshu_api = XiaohongshuSearchRecommendations()
+    query_suggestions = xiaohongshu_api.get_recommendations(keyword=query)
+    print(query_suggestions)
+    async def evaluate_single_query(q_sug: str, q_with_context: str):
+        """Evaluate a single query suggestion."""
+        eval_input = f"""
+{q_with_context}
+<待评估的推荐query>
+{q_sug}
+</待评估的推荐query>
+        """
+        evaluator_result = await Runner.run(evaluator, eval_input)
+        result: EvaluationFeedback = evaluator_result.final_output
+        return {
+            "query": q_sug,
+            "score": result.score,
+            "reason": result.reason,
+        }
+
+    # 并发执行所有评估任务
+    q_with_context = wrapper.context.q_with_context
+    res = []
+    if query_suggestions:
+        res = await asyncio.gather(*[evaluate_single_query(q_sug, q_with_context) for q_sug in query_suggestions])
+    else:
+        res = '未返回任何推荐词'
+
+    # 记录到 RunContext
+    wrapper.context.operations_history.append({
+        "operation_type": "get_query_suggestions",
+        "timestamp": datetime.now().isoformat(),
+        "query": query,
+        "suggestions": query_suggestions,
+        "evaluations": res,
+    })
+
+    return res
+
+@function_tool
+def modify_query(wrapper: RunContextWrapper[RunContext], original_query: str, operation_type: str, new_query: str, reason: str):
+    """
+    Modify the search query with a specific operation.
+
+    Args:
+        original_query: The original query before modification
+        operation_type: Type of modification - must be one of:
+            - "增加": 增加query语句，添加具体化的关键词
+            - "删减": 删减query词，去除冗余词汇
+            - "调序": 调整query词的顺序，更符合用户习惯的词序
+            - "替换": 替换query中的某个词，使用同义词或专业术语
+        new_query: The modified query after applying the operation
+        reason: Detailed explanation of why this modification was made and what insight from previous suggestions led to this change.
+                Must reference specific data from get_query_suggestions results (e.g., scores, reasons, word patterns).
+
+    Returns:
+        A dict containing the modification record and the new query to use for next search
+    """
+    operation_types = ["增加", "删减", "调序", "替换"]
+    if operation_type not in operation_types:
+        return {
+            "status": "error",
+            "message": f"Invalid operation_type. Must be one of: {', '.join(operation_types)}"
+        }
+
+    modification_record = {
+        "original_query": original_query,
+        "operation_type": operation_type,
+        "new_query": new_query,
+        "reason": reason,
+    }
+
+    # 记录到 RunContext
+    wrapper.context.operations_history.append({
+        "operation_type": "modify_query",
+        "timestamp": datetime.now().isoformat(),
+        "modification_type": operation_type,
+        "original_query": original_query,
+        "new_query": new_query,
+        "reason": reason,
+    })
+
+    return {
+        "status": "success",
+        "modification_record": modification_record,
+        "new_query": new_query,
+        "message": f"Query modified successfully. Use '{new_query}' for the next search."
+    }
+
+insrtuctions = """
+你是一个专业的搜索query优化专家，擅长通过动态探索找到最符合用户搜索习惯的query。
+
+## 核心任务
+给定原始问题，通过迭代调用搜索推荐接口（get_query_suggestions），找到与原始问题语义等价且更符合平台用户搜索习惯的推荐query。
+
+## 重要说明
+- **你不需要自己评估query的等价性**
+- get_query_suggestions 函数内部已集成评估子agent，会自动对每个推荐词进行评估
+- 返回结果包含：query（推荐词）、score（评分，1表示等价，0表示不等价）、reason（评估理由）
+- **你的职责是分析评估结果，做出决策和策略调整**
+
+## 防止幻觉 - 关键原则
+- **严禁编造数据**：只能基于 get_query_suggestions 实际返回的结果进行分析
+- **空结果处理**：如果返回的列表为空（[]），必须明确说明"未返回任何推荐词"
+- **不要猜测**：在 modify_query 的 reason 中，不能引用不存在的推荐词或评分
+- **如实记录**：每次分析都要如实反映实际返回的数据
+
+## 工作流程
+
+### 1. 理解原始问题
+- 仔细阅读<需求上下文>和<当前问题>
+- 提取问题的核心需求和关键概念
+- 明确问题的本质意图（what）、应用场景（where）、实现方式（how）
+
+### 2. 动态探索策略
+
+**第一轮尝试：**
+- 使用原始问题直接调用 get_query_suggestions(query="原始问题")
+- **检查返回结果**：
+  - 如果返回空列表 []：说明"该query未返回任何推荐词"，需要简化或替换query
+  - 如果有推荐词：查看每个推荐词的 score 和 reason
+- **做出判断**：是否有 score >= 0.8 的高分推荐词？
+
+**后续迭代：**
+如果没有高分推荐词（或返回空列表），必须先调用 modify_query 记录修改，然后再次搜索：
+
+**工具使用流程：**
+1. **分析评估反馈**（必须基于实际返回的数据）：
+   - **情况A - 返回空列表**：
+     * 在 reason 中说明："第X轮未返回任何推荐词，可能是query过于复杂或生僻"
+     * 不能编造任何推荐词或评分
+   - **情况B - 有推荐词但无高分**：
+     * 哪些推荐词得分较高？具体是多少分？评估理由是什么？
+     * 哪些推荐词偏离了原问题？如何偏离的？
+     * 推荐词整体趋势是什么？（过于泛化/具体化/领域偏移等）
+
+2. **决策修改策略**：基于实际评估反馈，调用 modify_query(original_query, operation_type, new_query, reason)
+   - reason 必须引用具体的数据，不能编造
+
+3. 使用返回的 new_query 调用 get_query_suggestions
+
+4. 分析新的评估结果，如果仍不满足，重复步骤1-3
+
+**四种操作类型（operation_type）及其详细判定标准：**
+
+**策略1：增加**
+- **判定标准**：
+  * 有效sug列表中，sug_i 的语义比 Query 更具体，且词汇数量 > query 的词汇数量
+  * sug list 中的每个词是 Query 的更具体或细化版本，是能将 Query 具体化描述的关键新词
+  * sug_i 必须包含 Query 的大部分或所有核心关键词，且词汇数量都明显多于query
+  * sug_j 中存在 query 中不存在的关键新词集合（例如：特定平台名称、特定功能词）
+- **使用时机**：当推荐词整体趋势是添加更具体的限定词或场景描述时
+
+**策略2：删减**
+- **判定标准**（满足以下任一条件即可）：
+  * 有效sug列表中，大部分 sug 的词汇数量都明显少于 query
+  * 当sug列表为空时
+  * sug 词汇数量明显少于 query，且与 Query 的核心搜索意图一致
+  * 基于评估反馈，识别 query 中 1-2 个词为"非核心冗余词"，且删除后不影响核心搜索意图
+- **使用时机**：当原query过于冗长复杂，推荐词都倾向于使用更简洁的表达时
+
+**策略3：调序**
+- **判定标准**：
+  * 有效sug list中，存在多个 sug_i 与 query 拥有完全相同的核心关键词集合，只是词汇排列顺序不同
+  * sug_i 与 query 的词序不同，但 sug_i 的词序变化与 query 的核心语义一致
+  * sug_i 的词序更自然或更符合搜索习惯，且 sug_i 的词序频率高于 query 词的词序
+  * 多个 Sug_List 中的 sug_i 都倾向于使用与 Query 不同但意思一致的词序
+- **使用时机**：当推荐词与原query关键词相同但顺序不同，且新顺序更符合用户习惯时
+
+**策略4：替换**
+- **判定标准**：
+  * 存在 sug_i（来自有效suglist），其与 query 之间仅有 1-2 个核心词不同，但其他词均相同
+  * sug_i 作为整体查询词，其语义与 query 等价或更优
+  * sug_i 与 Query 在大部分词汇上重叠，但在 1-2 个关键位置使用了不同的词
+  * 这些不同的词在语义上是同义、近义或在相关领域中更流行/专业的替代词
+  * 差异词中的新词在 Platform Sug List 中出现频率更高，或被标记为更专业的术语
+- **使用时机**：当推荐词显示某个词的同义词或专业术语更受欢迎时
+
+**每次修改的reason必须包含：**
+- 上一轮评估结果的关键发现（引用具体的score和reason）
+- 基于评估反馈，为什么这样修改
+- 预期这次修改会带来什么改进
+
+### 3. 决策标准
+- **score >= 0.8**：认为该推荐词与原问题等价，可以作为最终结果
+- **0.5 <= score < 0.8**：部分等价，分析reason看是否可接受
+- **score < 0.5**：不等价，需要继续优化
+
+### 4. 迭代终止条件
+- **成功终止**：找到至少一个 score >= 0.8 的推荐query
+- **尝试上限**：最多迭代5轮，避免无限循环
+- **无推荐词**：推荐接口返回空列表或错误
+
+### 5. 输出要求
+
+**成功找到等价query时，输出格式：**
+```
+原始问题：[原问题]
+优化后的query：[最终找到的等价推荐query]
+评分：[score]
+```
+
+**未找到等价query时，输出格式：**
+```
+原始问题：[原问题]
+结果：未找到完全等价的推荐query
+建议：[简要建议，如：直接使用原问题搜索 或 使用最接近的推荐词]
+```
+
+## 注意事项
+- **第一轮必须使用原始问题**：直接调用 get_query_suggestions(query="原始问题")
+- **后续修改必须调用 modify_query**：不能直接用新query调用 get_query_suggestions
+- **重点关注评估结果**：每次都要仔细分析返回的 score 和 reason
+- **基于数据决策**：修改策略必须基于评估反馈，不能凭空猜测
+- **引用具体评分**：在分析和决策时，引用具体的score数值和reason内容
+- **优先选择高分推荐词**：score >= 0.8 即可认为等价
+- **严禁编造数据**：
+  * 如果返回空列表，必须在 reason 中明确说明"未返回任何推荐词"
+  * 不能引用不存在的推荐词、评分或评估理由
+  * 每次 modify_query 的 reason 必须基于上一轮实际返回的结果
+""".strip()
+
+
+
+
+async def main():
+    current_time, log_url = set_trace()
+    q_context = read_file_as_string('input/kg_v1_single_context.md')
+    q = read_file_as_string('input/kg_v1_single_q.md')
+    q_with_context = f"""
+<需求上下文>
+{q_context}
+</需求上下文>
+<当前问题>
+{q}
+</当前问题>
+""".strip()
+    log_dir = os.path.join("logs", current_time)
+    run_context = RunContext(
+        q_with_context=q_with_context,
+        q_context=q_context,
+        q=q,
+        log_dir=log_dir,
+        log_url=log_url,
+    )
+    agent = Agent[RunContext](
+        name="Query Optimization Agent",
+        instructions=insrtuctions,
+        tools=[get_query_suggestions, modify_query],
+       
+    )
+    result = await Runner.run(agent, input=q_with_context,  context = run_context,)
+    print(result.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}")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

sug_v2_1_0.py

@@ -0,0 +1,307 @@
+import asyncio
+import json
+import os
+from datetime import datetime
+
+from agents import Agent, Runner, function_tool
+from lib.my_trace import set_trace
+from typing import Literal
+from dataclasses import dataclass
+from pydantic import BaseModel, Field
+
+
+from lib.utils import read_file_as_string
+from script.search_recommendations.xiaohongshu_search_recommendations import XiaohongshuSearchRecommendations
+
+from agents import Agent, RunContextWrapper, Runner, function_tool
+
+from pydantic import BaseModel, Field
+
+
+class RunContext(BaseModel):
+    q_with_context: str
+    q_context: str
+    q: str
+    log_url: str
+    log_dir: str
+    # 中间数据记录 - 按时间顺序记录所有操作
+    operations_history: list[dict] = Field(default_factory=list, description="记录所有操作的历史，包括 get_query_suggestions 和 modify_query")
+
+
+eval_insrtuctions = """
+你是一个专业的评估专家，负责评估给定的搜索query是否满足原始问题和需求，给出出评分和简明扼要的理由。
+"""
+
+@dataclass
+class EvaluationFeedback:
+    reason: str=Field(..., description="简明扼要的理由")
+    score: float=Field(..., description="评估结果，1表示等价，0表示不等价，中间值表示部分等价")
+
+evaluator = Agent[None](
+    name="评估专家",
+    instructions=eval_insrtuctions,
+    output_type=EvaluationFeedback,
+)
+
+@function_tool
+async def get_query_suggestions(wrapper: RunContextWrapper[RunContext], query: str):
+    """Fetch search recommendations from Xiaohongshu."""
+    xiaohongshu_api = XiaohongshuSearchRecommendations()
+    query_suggestions = xiaohongshu_api.get_recommendations(keyword=query)
+    print(query_suggestions)
+    async def evaluate_single_query(q_sug: str, q_with_context: str):
+        """Evaluate a single query suggestion."""
+        eval_input = f"""
+{q_with_context}
+<待评估的推荐query>
+{q_sug}
+</待评估的推荐query>
+        """
+        evaluator_result = await Runner.run(evaluator, eval_input)
+        result: EvaluationFeedback = evaluator_result.final_output
+        return {
+            "query": q_sug,
+            "score": result.score,
+            "reason": result.reason,
+        }
+
+    # 并发执行所有评估任务
+    q_with_context = wrapper.context.q_with_context
+    res = await asyncio.gather(*[evaluate_single_query(q_sug, q_with_context) for q_sug in query_suggestions])
+
+    # 记录到 RunContext
+    wrapper.context.operations_history.append({
+        "operation_type": "get_query_suggestions",
+        "timestamp": datetime.now().isoformat(),
+        "query": query,
+        "suggestions": query_suggestions,
+        "evaluations": res,
+    })
+
+    return res
+
+@function_tool
+def modify_query(wrapper: RunContextWrapper[RunContext], original_query: str, operation_type: str, new_query: str, reason: str):
+    """
+    Modify the search query with a specific operation.
+
+    Args:
+        original_query: The original query before modification
+        operation_type: Type of modification - must be one of:
+            - "简化": 删减query词，去除冗余词汇
+            - "扩展": 增加query语句，添加具体化的关键词
+            - "替换": 替换query中的某个词，使用同义词或专业术语
+            - "组合": 更换query词的顺序，调整为更符合用户习惯的词序
+        new_query: The modified query after applying the operation
+        reason: Detailed explanation of why this modification was made and what insight from previous suggestions led to this change.
+                Must reference specific data from get_query_suggestions results (e.g., scores, reasons, word patterns).
+
+    Returns:
+        A dict containing the modification record and the new query to use for next search
+    """
+    operation_types = ["简化", "扩展", "替换", "组合"]
+    if operation_type not in operation_types:
+        return {
+            "status": "error",
+            "message": f"Invalid operation_type. Must be one of: {', '.join(operation_types)}"
+        }
+
+    modification_record = {
+        "original_query": original_query,
+        "operation_type": operation_type,
+        "new_query": new_query,
+        "reason": reason,
+    }
+
+    # 记录到 RunContext
+    wrapper.context.operations_history.append({
+        "operation_type": "modify_query",
+        "timestamp": datetime.now().isoformat(),
+        "modification_type": operation_type,
+        "original_query": original_query,
+        "new_query": new_query,
+        "reason": reason,
+    })
+
+    return {
+        "status": "success",
+        "modification_record": modification_record,
+        "new_query": new_query,
+        "message": f"Query modified successfully. Use '{new_query}' for the next search."
+    }
+
+insrtuctions = """
+你是一个专业的搜索query优化专家，擅长通过动态探索找到最符合用户搜索习惯的query。
+
+## 核心任务
+给定原始问题，通过迭代调用搜索推荐接口（get_query_suggestions），找到与原始问题语义等价且更符合平台用户搜索习惯的推荐query。
+
+## 重要说明
+- **你不需要自己评估query的等价性**
+- get_query_suggestions 函数内部已集成评估子agent，会自动对每个推荐词进行评估
+- 返回结果包含：query（推荐词）、score（评分，1表示等价，0表示不等价）、reason（评估理由）
+- **你的职责是分析评估结果，做出决策和策略调整**
+
+## 防止幻觉 - 关键原则
+- **严禁编造数据**：只能基于 get_query_suggestions 实际返回的结果进行分析
+- **空结果处理**：如果返回的列表为空（[]），必须明确说明"未返回任何推荐词"
+- **不要猜测**：在 modify_query 的 reason 中，不能引用不存在的推荐词或评分
+- **如实记录**：每次分析都要如实反映实际返回的数据
+
+## 工作流程
+
+### 1. 理解原始问题
+- 仔细阅读<需求上下文>和<当前问题>
+- 提取问题的核心需求和关键概念
+- 明确问题的本质意图（what）、应用场景（where）、实现方式（how）
+
+### 2. 动态探索策略
+
+**第一轮尝试：**
+- 使用原始问题直接调用 get_query_suggestions(query="原始问题")
+- **检查返回结果**：
+  - 如果返回空列表 []：说明"该query未返回任何推荐词"，需要简化或替换query
+  - 如果有推荐词：查看每个推荐词的 score 和 reason
+- **做出判断**：是否有 score >= 0.8 的高分推荐词？
+
+**后续迭代：**
+如果没有高分推荐词（或返回空列表），必须先调用 modify_query 记录修改，然后再次搜索：
+
+**工具使用流程：**
+1. **分析评估反馈**（必须基于实际返回的数据）：
+   - **情况A - 返回空列表**：
+     * 在 reason 中说明："第X轮未返回任何推荐词，可能是query过于复杂或生僻"
+     * 不能编造任何推荐词或评分
+   - **情况B - 有推荐词但无高分**：
+     * 哪些推荐词得分较高？具体是多少分？评估理由是什么？
+     * 哪些推荐词偏离了原问题？如何偏离的？
+     * 推荐词整体趋势是什么？（过于泛化/具体化/领域偏移等）
+
+2. **决策修改策略**：基于实际评估反馈，调用 modify_query(original_query, operation_type, new_query, reason)
+   - reason 必须引用具体的数据，不能编造
+
+3. 使用返回的 new_query 调用 get_query_suggestions
+
+4. 分析新的评估结果，如果仍不满足，重复步骤1-3
+
+**四种操作类型（operation_type）及其详细判定标准：**
+
+**策略1：扩展（增加 query 语句）**
+- **判定标准**：
+  * 有效sug列表中，sug_i 的语义比 Query 更具体，且词汇数量 > query 的词汇数量
+  * sug list 中的每个词是 Query 的更具体或细化版本，是能将 Query 具体化描述的关键新词
+  * sug_i 必须包含 Query 的大部分或所有核心关键词，且词汇数量都明显多于query
+  * sug_j 中存在 query 中不存在的关键新词集合（例如：特定平台名称、特定功能词）
+- **使用时机**：当推荐词整体趋势是添加更具体的限定词或场景描述时
+
+**策略2：简化（删减 query 词）**
+- **判定标准**（满足以下任一条件即可）：
+  * 有效sug列表中，大部分 sug 的词汇数量都明显少于 query
+  * 当sug列表为空时
+  * sug 词汇数量明显少于 query，且与 Query 的核心搜索意图一致
+  * 基于评估反馈，识别 query 中 1-2 个词为"非核心冗余词"，且删除后不影响核心搜索意图
+- **使用时机**：当原query过于冗长复杂，推荐词都倾向于使用更简洁的表达时
+
+**策略3：组合（更换 query 词的顺序）**
+- **判定标准**：
+  * 有效sug list中，存在多个 sug_i 与 query 拥有完全相同的核心关键词集合，只是词汇排列顺序不同
+  * sug_i 与 query 的词序不同，但 sug_i 的词序变化与 query 的核心语义一致
+  * sug_i 的词序更自然或更符合搜索习惯，且 sug_i 的词序频率高于 query 词的词序
+  * 多个 Sug_List 中的 sug_i 都倾向于使用与 Query 不同但意思一致的词序
+- **使用时机**：当推荐词与原query关键词相同但顺序不同，且新顺序更符合用户习惯时
+
+**策略4：替换（替换 query 语句中的某个词）**
+- **判定标准**：
+  * 存在 sug_i（来自有效suglist），其与 query 之间仅有 1-2 个核心词不同，但其他词均相同
+  * sug_i 作为整体查询词，其语义与 query 等价或更优
+  * sug_i 与 Query 在大部分词汇上重叠，但在 1-2 个关键位置使用了不同的词
+  * 这些不同的词在语义上是同义、近义或在相关领域中更流行/专业的替代词
+  * 差异词中的新词在 Platform Sug List 中出现频率更高，或被标记为更专业的术语
+- **使用时机**：当推荐词显示某个词的同义词或专业术语更受欢迎时
+
+**每次修改的reason必须包含：**
+- 上一轮评估结果的关键发现（引用具体的score和reason）
+- 基于评估反馈，为什么这样修改
+- 预期这次修改会带来什么改进
+
+### 3. 决策标准
+- **score >= 0.8**：认为该推荐词与原问题等价，可以作为最终结果
+- **0.5 <= score < 0.8**：部分等价，分析reason看是否可接受
+- **score < 0.5**：不等价，需要继续优化
+
+### 4. 迭代终止条件
+- **成功终止**：找到至少一个 score >= 0.8 的推荐query
+- **尝试上限**：最多迭代5轮，避免无限循环
+- **无推荐词**：推荐接口返回空列表或错误
+
+### 5. 输出要求
+
+**成功找到等价query时，输出格式：**
+```
+原始问题：[原问题]
+优化后的query：[最终找到的等价推荐query]
+评分：[score]
+```
+
+**未找到等价query时，输出格式：**
+```
+原始问题：[原问题]
+结果：未找到完全等价的推荐query
+建议：[简要建议，如：直接使用原问题搜索 或 使用最接近的推荐词]
+```
+
+## 注意事项
+- **第一轮必须使用原始问题**：直接调用 get_query_suggestions(query="原始问题")
+- **后续修改必须调用 modify_query**：不能直接用新query调用 get_query_suggestions
+- **重点关注评估结果**：每次都要仔细分析返回的 score 和 reason
+- **基于数据决策**：修改策略必须基于评估反馈，不能凭空猜测
+- **引用具体评分**：在分析和决策时，引用具体的score数值和reason内容
+- **优先选择高分推荐词**：score >= 0.8 即可认为等价
+- **严禁编造数据**：
+  * 如果返回空列表，必须在 reason 中明确说明"未返回任何推荐词"
+  * 不能引用不存在的推荐词、评分或评估理由
+  * 每次 modify_query 的 reason 必须基于上一轮实际返回的结果
+""".strip()
+
+
+
+
+async def main():
+    current_time, log_url = set_trace()
+    q_context = read_file_as_string('input/kg_v1_single_context.md')
+    q = read_file_as_string('input/kg_v1_single_q.md')
+    q_with_context = f"""
+<需求上下文>
+{q_context}
+</需求上下文>
+<当前问题>
+{q}
+</当前问题>
+""".strip()
+    log_dir = os.path.join("logs", current_time)
+    run_context = RunContext(
+        q_with_context=q_with_context,
+        q_context=q_context,
+        q=q,
+        log_dir=log_dir,
+        log_url=log_url,
+    )
+    agent = Agent[RunContext](
+        name="Query Optimization Agent",
+        instructions=insrtuctions,
+        tools=[get_query_suggestions, modify_query],
+       
+    )
+    result = await Runner.run(agent, input=q_with_context,  context = run_context,)
+    print(result.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}")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

sug_v2_with_eval_yx.py

@@ -0,0 +1,349 @@
+import asyncio
+import json
+import os
+import argparse
+from datetime import datetime
+
+from agents import Agent, Runner, function_tool
+from lib.my_trace import set_trace
+from typing import Literal
+from dataclasses import dataclass
+from pydantic import BaseModel, Field
+
+
+from lib.utils import read_file_as_string
+from script.search_recommendations.xiaohongshu_search_recommendations import XiaohongshuSearchRecommendations
+
+from agents import Agent, RunContextWrapper, Runner, function_tool
+
+from pydantic import BaseModel, Field
+
+
+class RunContext(BaseModel):
+    version: str = Field(..., description="当前运行的脚本版本（文件名）")
+    input_files: dict[str, str] = Field(..., description="输入文件路径映射，如 {'context_file': '...', 'q_file': '...'}")
+    q_with_context: str
+    q_context: str
+    q: str
+    log_url: str
+    log_dir: str
+    # 中间数据记录 - 按时间顺序记录所有操作
+    operations_history: list[dict] = Field(default_factory=list, description="记录所有操作的历史，包括 get_query_suggestions 和 modify_query")
+    # 最终输出结果
+    final_output: str | None = Field(default=None, description="Agent的最终输出结果")
+
+
+eval_insrtuctions = """
+# 角色定义
+
+你是一个 **专业的语言专家和语义相关性评判专家**。
+
+你的任务是：判断我给你的平台sug词条，其与 <原始 query 问题> 和 <需求上下文> **共同形成的综合意图** 的相关度满足度百分比。
+
+---
+
+# 输入信息
+
+你将接收到以下输入：
+1.  <需求上下文>：文档的解构结果或语义说明，用于理解 query 所针对的核心任务与场景。
+2.  <原始 query 问题>：用户的初始查询问题。
+3.  <平台sug词条>：模型或算法推荐的 平台sug词条，其中包含待评估的词条。
+
+---
+
+# 工作流程
+
+## 第一步：理解综合意图
+
+-   仔细阅读 <需求上下文> 与 <原始 query 问题>；
+-   明确 <原始 query 问题> 的核心意图、主题焦点和任务目标；
+-   提炼 <需求上下文> 所描述的核心任务、场景、概念和语义范围；
+-   **将 <原始 query 问题> 和 <需求上下文> 融合成一个单一的、完整的“综合意图”**，作为后续评估的唯一基准。这个综合意图代表了用户在特定场景下最核心、最全面的需求。
+
+## 第二步：逐一评估 Sug 词条的综合相关度满足度
+
+对于 我给你的sug词条：
+
+1.  **评估其与“综合意图”的相关度满足度：**
+    -   将当前 sug 词条的语义与第一步中确定的“综合意图”（<原始 query 问题> 和 <需求上下文> 的结合体）进行比较。
+    -   全面评估该 sug 词条在主题、目标动作、语义范围、场景匹配、核心概念覆盖等各方面与“综合意图”的匹配和满足程度。
+    -   给出 0-1 之间的浮点数，代表其“相关度满足度”。数值比越高，表示该 sug 词条越能充分且精准地满足“综合意图”。
+
+**评估标准一致性要求：在评估 sug 词条时，请务必保持你对“相关度满足度”的判断标准和百分比给定的逻辑完全一致和稳定。**
+
+**相关性依据要求：** 简短叙述相关性依据，说明主要匹配点或差异点，以及该 sug 词条如何满足或未能完全满足“综合意图”的哪些方面。
+"""
+
+@dataclass
+class EvaluationFeedback:
+    reason: str=Field(..., description="简明扼要的理由")
+    score: float=Field(..., description="评估结果，1表示等价，0表示不等价，中间值表示部分等价")
+
+evaluator = Agent[None](
+    name="评估专家",
+    instructions=eval_insrtuctions,
+    output_type=EvaluationFeedback,
+)
+
+@function_tool
+async def get_query_suggestions(wrapper: RunContextWrapper[RunContext], query: str):
+    """Fetch search recommendations from Xiaohongshu."""
+    xiaohongshu_api = XiaohongshuSearchRecommendations()
+    query_suggestions = xiaohongshu_api.get_recommendations(keyword=query)
+    print(query_suggestions)
+    async def evaluate_single_query(q_sug: str, q_with_context: str):
+        """Evaluate a single query suggestion."""
+        eval_input = f"""
+{q_with_context}
+<待评估的推荐query>
+{q_sug}
+</待评估的推荐query>
+        """
+        evaluator_result = await Runner.run(evaluator, eval_input)
+        result: EvaluationFeedback = evaluator_result.final_output
+        return {
+            "query": q_sug,
+            "score": result.score,
+            "reason": result.reason,
+        }
+
+    # 并发执行所有评估任务
+    q_with_context = wrapper.context.q_with_context
+    res = []
+    if query_suggestions:
+        res = await asyncio.gather(*[evaluate_single_query(q_sug, q_with_context) for q_sug in query_suggestions])
+    else:
+        res = '未返回任何推荐词'
+
+    # 记录到 RunContext
+    wrapper.context.operations_history.append({
+        "operation_type": "get_query_suggestions",
+        "timestamp": datetime.now().isoformat(),
+        "query": query,
+        "suggestions": query_suggestions,
+        "evaluations": res,
+    })
+
+    return res
+
+@function_tool
+def modify_query(wrapper: RunContextWrapper[RunContext], original_query: str, operation_type: str, new_query: str, reason: str):
+    """
+    Modify the search query with a specific operation.
+
+    Args:
+        original_query: The original query before modification
+        operation_type: Type of modification - must be one of: "简化", "扩展", "替换", "组合"
+        new_query: The modified query after applying the operation
+        reason: Detailed explanation of why this modification was made and what insight from previous suggestions led to this change
+
+    Returns:
+        A dict containing the modification record and the new query to use for next search
+    """
+    operation_types = ["简化", "扩展", "替换", "组合"]
+    if operation_type not in operation_types:
+        return {
+            "status": "error",
+            "message": f"Invalid operation_type. Must be one of: {', '.join(operation_types)}"
+        }
+
+    modification_record = {
+        "original_query": original_query,
+        "operation_type": operation_type,
+        "new_query": new_query,
+        "reason": reason,
+    }
+
+    # 记录到 RunContext
+    wrapper.context.operations_history.append({
+        "operation_type": "modify_query",
+        "timestamp": datetime.now().isoformat(),
+        "modification_type": operation_type,
+        "original_query": original_query,
+        "new_query": new_query,
+        "reason": reason,
+    })
+
+    return {
+        "status": "success",
+        "modification_record": modification_record,
+        "new_query": new_query,
+        "message": f"Query modified successfully. Use '{new_query}' for the next search."
+    }
+
+insrtuctions = """
+你是一个专业的搜索query优化专家，擅长通过动态探索找到最符合用户搜索习惯的query。
+
+## 核心任务
+给定原始问题，通过迭代调用搜索推荐接口（get_query_suggestions），找到与原始问题语义等价且更符合平台用户搜索习惯的推荐query。
+
+## 重要说明
+- **你不需要自己评估query的等价性**
+- get_query_suggestions 函数内部已集成评估子agent，会自动对每个推荐词进行评估
+- 返回结果包含：query（推荐词）、score（评分，1表示等价，0表示不等价）、reason（评估理由）
+- **你的职责是分析评估结果，做出决策和策略调整**
+
+## 防止幻觉 - 关键原则
+- **严禁编造数据**：只能基于 get_query_suggestions 实际返回的结果进行分析
+- **空结果处理**：如果返回的列表为空（[]），必须明确说明"未返回任何推荐词"
+- **不要猜测**：在 modify_query 的 reason 中，不能引用不存在的推荐词或评分
+- **如实记录**：每次分析都要如实反映实际返回的数据
+
+## 工作流程
+
+### 1. 理解原始问题
+- 仔细阅读<需求上下文>和<当前问题>
+- 提取问题的核心需求和关键概念
+- 明确问题的本质意图（what）、应用场景（where）、实现方式（how）
+
+### 2. 动态探索策略
+
+**第一轮尝试：**
+- 使用原始问题直接调用 get_query_suggestions(query="原始问题")
+- **检查返回结果**：
+  - 如果返回空列表 []：说明"该query未返回任何推荐词"，需要简化或替换query
+  - 如果有推荐词：查看每个推荐词的 score 和 reason
+- **做出判断**：是否有 score >= 0.8 的高分推荐词？
+
+**后续迭代：**
+如果没有高分推荐词（或返回空列表），必须先调用 modify_query 记录修改，然后再次搜索：
+
+**工具使用流程：**
+1. **分析评估反馈**（必须基于实际返回的数据）：
+   - **情况A - 返回空列表**：
+     * 在 reason 中说明："第X轮未返回任何推荐词，可能是query过于复杂或生僻"
+     * 不能编造任何推荐词或评分
+   - **情况B - 有推荐词但无高分**：
+     * 哪些推荐词得分较高？具体是多少分？评估理由是什么？
+     * 哪些推荐词偏离了原问题？如何偏离的？
+     * 推荐词整体趋势是什么？（过于泛化/具体化/领域偏移等）
+
+2. **决策修改策略**：基于实际评估反馈，调用 modify_query(original_query, operation_type, new_query, reason)
+   - reason 必须引用具体的数据，不能编造
+
+3. 使用返回的 new_query 调用 get_query_suggestions
+
+4. 分析新的评估结果，如果仍不满足，重复步骤1-3
+
+**四种操作类型（operation_type）：**
+- **简化**：删除冗余词汇，提取核心关键词（当推荐词过于发散时）
+- **扩展**：添加限定词或场景描述（当推荐词过于泛化时）
+- **替换**：使用同义词、行业术语或口语化表达（当推荐词偏离核心时）
+- **组合**：调整关键词顺序或组合方式（当推荐词结构不合理时）
+
+**每次修改的reason必须包含：**
+- 上一轮评估结果的关键发现（引用具体的score和reason）
+- 基于评估反馈，为什么这样修改
+- 预期这次修改会带来什么改进
+
+### 3. 决策标准
+- **score >= 0.8**：认为该推荐词与原问题等价，可以作为最终结果
+- **0.5 <= score < 0.8**：部分等价，分析reason看是否可接受
+- **score < 0.5**：不等价，需要继续优化
+
+### 4. 迭代终止条件
+- **成功终止**：找到至少一个 score >= 0.8 的推荐query
+- **尝试上限**：最多迭代5轮，避免无限循环
+- **无推荐词**：推荐接口返回空列表或错误
+
+### 5. 输出要求
+
+**成功找到等价query时，输出格式：**
+```
+原始问题：[原问题]
+优化后的query：[最终找到的等价推荐query]
+评分：[score]
+```
+
+**未找到等价query时，输出格式：**
+```
+原始问题：[原问题]
+结果：未找到完全等价的推荐query
+建议：[简要建议，如：直接使用原问题搜索 或 使用最接近的推荐词]
+```
+
+## 注意事项
+- **第一轮必须使用原始问题**：直接调用 get_query_suggestions(query="原始问题")
+- **后续修改必须调用 modify_query**：不能直接用新query调用 get_query_suggestions
+- **重点关注评估结果**：每次都要仔细分析返回的 score 和 reason
+- **基于数据决策**：修改策略必须基于评估反馈，不能凭空猜测
+- **引用具体评分**：在分析和决策时，引用具体的score数值和reason内容
+- **优先选择高分推荐词**：score >= 0.8 即可认为等价
+- **严禁编造数据**：
+  * 如果返回空列表，必须在 reason 中明确说明"未返回任何推荐词"
+  * 不能引用不存在的推荐词、评分或评估理由
+  * 每次 modify_query 的 reason 必须基于上一轮实际返回的结果
+""".strip()
+
+
+
+
+async def main(input_dir: str):
+    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]  # 去掉 .py 后缀
+
+    # 日志保存到输入目录的 output/版本/时间戳 目录下
+    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,
+    )
+    agent = Agent[RunContext](
+        name="Query Optimization Agent",
+        instructions=insrtuctions,
+        tools=[get_query_suggestions, modify_query],
+       
+    )
+    result = await Runner.run(agent, input=q_with_context,  context = run_context,)
+    print(result.final_output)
+
+    # 保存最终输出到 RunContext
+    run_context.final_output = str(result.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}")
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description="搜索query优化工具")
+    parser.add_argument(
+        "--input-dir",
+        type=str,
+        default="input/简单扣图",
+        help="输入目录路径，默认: input/简单扣图"
+    )
+    args = parser.parse_args()
+
+    asyncio.run(main(args.input_dir))

sug_v3.py

@@ -0,0 +1,312 @@
+import asyncio
+import json
+import os
+import argparse
+from datetime import datetime
+
+from agents import Agent, Runner, function_tool
+from lib.my_trace import set_trace
+from typing import Literal
+from dataclasses import dataclass
+from pydantic import BaseModel, Field
+
+
+from lib.utils import read_file_as_string
+from script.search_recommendations.xiaohongshu_search_recommendations import XiaohongshuSearchRecommendations
+
+from agents import Agent, RunContextWrapper, Runner, function_tool
+
+from pydantic import BaseModel, Field
+
+
+class RunContext(BaseModel):
+    version: str = Field(..., description="当前运行的脚本版本（文件名）")
+    input_files: dict[str, str] = Field(..., description="输入文件路径映射，如 {'context_file': '...', 'q_file': '...'}")
+    q_with_context: str
+    q_context: str
+    q: str
+    log_url: str
+    log_dir: str
+    # 中间数据记录 - 按时间顺序记录所有操作
+    operations_history: list[dict] = Field(default_factory=list, description="记录所有操作的历史，包括 get_query_suggestions 和 modify_query")
+    # 最终输出结果
+    final_output: str | None = Field(default=None, description="Agent的最终输出结果")
+
+
+eval_insrtuctions = """
+你是一个专业的评估专家，负责评估给定的搜索query是否满足原始问题和需求，给出出评分和简明扼要的理由。
+"""
+
+@dataclass
+class EvaluationFeedback:
+    reason: str=Field(..., description="简明扼要的理由")
+    score: float=Field(..., description="评估结果，1表示等价，0表示不等价，中间值表示部分等价")
+
+evaluator = Agent[None](
+    name="评估专家",
+    instructions=eval_insrtuctions,
+    output_type=EvaluationFeedback,
+)
+
+@function_tool
+async def get_query_suggestions(wrapper: RunContextWrapper[RunContext], query: str):
+    """Fetch search recommendations from Xiaohongshu."""
+    xiaohongshu_api = XiaohongshuSearchRecommendations()
+    query_suggestions = xiaohongshu_api.get_recommendations(keyword=query)
+    print(query_suggestions)
+    async def evaluate_single_query(q_sug: str, q_with_context: str):
+        """Evaluate a single query suggestion."""
+        eval_input = f"""
+{q_with_context}
+<待评估的推荐query>
+{q_sug}
+</待评估的推荐query>
+        """
+        evaluator_result = await Runner.run(evaluator, eval_input)
+        result: EvaluationFeedback = evaluator_result.final_output
+        return {
+            "query": q_sug,
+            "score": result.score,
+            "reason": result.reason,
+        }
+
+    # 并发执行所有评估任务
+    q_with_context = wrapper.context.q_with_context
+    res = []
+    if query_suggestions:
+        res = await asyncio.gather(*[evaluate_single_query(q_sug, q_with_context) for q_sug in query_suggestions])
+    else:
+        res = '未返回任何推荐词'
+
+    # 记录到 RunContext
+    wrapper.context.operations_history.append({
+        "operation_type": "get_query_suggestions",
+        "timestamp": datetime.now().isoformat(),
+        "query": query,
+        "suggestions": query_suggestions,
+        "evaluations": res,
+    })
+
+    return res
+
+@function_tool
+def modify_query(wrapper: RunContextWrapper[RunContext], original_query: str, operation_type: str, new_query: str, reason: str):
+    """
+    Modify the search query with a specific operation.
+
+    Args:
+        original_query: The original query before modification
+        operation_type: Type of modification - must be one of: "简化", "扩展", "替换", "组合"
+        new_query: The modified query after applying the operation
+        reason: Detailed explanation of why this modification was made and what insight from previous suggestions led to this change
+
+    Returns:
+        A dict containing the modification record and the new query to use for next search
+    """
+    operation_types = ["简化", "扩展", "替换", "组合"]
+    if operation_type not in operation_types:
+        return {
+            "status": "error",
+            "message": f"Invalid operation_type. Must be one of: {', '.join(operation_types)}"
+        }
+
+    modification_record = {
+        "original_query": original_query,
+        "operation_type": operation_type,
+        "new_query": new_query,
+        "reason": reason,
+    }
+
+    # 记录到 RunContext
+    wrapper.context.operations_history.append({
+        "operation_type": "modify_query",
+        "timestamp": datetime.now().isoformat(),
+        "modification_type": operation_type,
+        "original_query": original_query,
+        "new_query": new_query,
+        "reason": reason,
+    })
+
+    return {
+        "status": "success",
+        "modification_record": modification_record,
+        "new_query": new_query,
+        "message": f"Query modified successfully. Use '{new_query}' for the next search."
+    }
+
+insrtuctions = """
+你是一个专业的搜索query优化专家，擅长通过动态探索找到最符合用户搜索习惯的query。
+
+## 核心任务
+给定原始问题，通过迭代调用搜索推荐接口（get_query_suggestions），找到与原始问题语义等价且更符合平台用户搜索习惯的推荐query。
+
+## 重要说明
+- **你不需要自己评估query的等价性**
+- get_query_suggestions 函数内部已集成评估子agent，会自动对每个推荐词进行评估
+- 返回结果包含：query（推荐词）、score（评分，1表示等价，0表示不等价）、reason（评估理由）
+- **你的职责是分析评估结果，做出决策和策略调整**
+
+## 防止幻觉 - 关键原则
+- **严禁编造数据**：只能基于 get_query_suggestions 实际返回的结果进行分析
+- **空结果处理**：如果返回的列表为空（[]），必须明确说明"未返回任何推荐词"
+- **不要猜测**：在 modify_query 的 reason 中，不能引用不存在的推荐词或评分
+- **如实记录**：每次分析都要如实反映实际返回的数据
+
+## 工作流程
+
+### 1. 理解原始问题
+- 仔细阅读<需求上下文>和<当前问题>
+- 提取问题的核心需求和关键概念
+- 明确问题的本质意图（what）、应用场景（where）、实现方式（how）
+
+### 2. 动态探索策略
+
+**第一轮尝试：**
+- 使用原始问题直接调用 get_query_suggestions(query="原始问题")
+- **检查返回结果**：
+  - 如果返回空列表 []：说明"该query未返回任何推荐词"，需要简化或替换query
+  - 如果有推荐词：查看每个推荐词的 score 和 reason
+- **做出判断**：是否有 score >= 0.8 的高分推荐词？
+
+**后续迭代：**
+如果没有高分推荐词（或返回空列表），必须先调用 modify_query 记录修改，然后再次搜索：
+
+**工具使用流程：**
+1. **分析评估反馈**（必须基于实际返回的数据）：
+   - **情况A - 返回空列表**：
+     * 在 reason 中说明："第X轮未返回任何推荐词，可能是query过于复杂或生僻"
+     * 不能编造任何推荐词或评分
+   - **情况B - 有推荐词但无高分**：
+     * 哪些推荐词得分较高？具体是多少分？评估理由是什么？
+     * 哪些推荐词偏离了原问题？如何偏离的？
+     * 推荐词整体趋势是什么？（过于泛化/具体化/领域偏移等）
+
+2. **决策修改策略**：基于实际评估反馈，调用 modify_query(original_query, operation_type, new_query, reason)
+   - reason 必须引用具体的数据，不能编造
+
+3. 使用返回的 new_query 调用 get_query_suggestions
+
+4. 分析新的评估结果，如果仍不满足，重复步骤1-3
+
+**四种操作类型（operation_type）：**
+- **简化**：删除冗余词汇，提取核心关键词（当推荐词过于发散时）
+- **扩展**：添加限定词或场景描述（当推荐词过于泛化时）
+- **替换**：使用同义词、行业术语或口语化表达（当推荐词偏离核心时）
+- **组合**：调整关键词顺序或组合方式（当推荐词结构不合理时）
+
+**每次修改的reason必须包含：**
+- 上一轮评估结果的关键发现（引用具体的score和reason）
+- 基于评估反馈，为什么这样修改
+- 预期这次修改会带来什么改进
+
+### 3. 决策标准
+- **score >= 0.8**：认为该推荐词与原问题等价，可以作为最终结果
+- **0.5 <= score < 0.8**：部分等价，分析reason看是否可接受
+- **score < 0.5**：不等价，需要继续优化
+
+### 4. 迭代终止条件
+- **成功终止**：找到至少一个 score >= 0.8 的推荐query
+- **尝试上限**：最多迭代5轮，避免无限循环
+- **无推荐词**：推荐接口返回空列表或错误
+
+### 5. 输出要求
+
+**成功找到等价query时，输出格式：**
+```
+原始问题：[原问题]
+优化后的query：[最终找到的等价推荐query]
+评分：[score]
+```
+
+**未找到等价query时，输出格式：**
+```
+原始问题：[原问题]
+结果：未找到完全等价的推荐query
+建议：[简要建议，如：直接使用原问题搜索 或 使用最接近的推荐词]
+```
+
+## 注意事项
+- **第一轮必须使用原始问题**：直接调用 get_query_suggestions(query="原始问题")
+- **后续修改必须调用 modify_query**：不能直接用新query调用 get_query_suggestions
+- **重点关注评估结果**：每次都要仔细分析返回的 score 和 reason
+- **基于数据决策**：修改策略必须基于评估反馈，不能凭空猜测
+- **引用具体评分**：在分析和决策时，引用具体的score数值和reason内容
+- **优先选择高分推荐词**：score >= 0.8 即可认为等价
+- **严禁编造数据**：
+  * 如果返回空列表，必须在 reason 中明确说明"未返回任何推荐词"
+  * 不能引用不存在的推荐词、评分或评估理由
+  * 每次 modify_query 的 reason 必须基于上一轮实际返回的结果
+""".strip()
+
+
+
+
+async def main(input_dir: str):
+    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]  # 去掉 .py 后缀
+
+    # 日志保存到输入目录的 output/版本/时间戳 目录下
+    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,
+    )
+    agent = Agent[RunContext](
+        name="Query Optimization Agent",
+        instructions=insrtuctions,
+        tools=[get_query_suggestions, modify_query],
+       
+    )
+    result = await Runner.run(agent, input=q_with_context,  context = run_context,)
+    print(result.final_output)
+
+    # 保存最终输出到 RunContext
+    run_context.final_output = str(result.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}")
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description="搜索query优化工具")
+    parser.add_argument(
+        "--input-dir",
+        type=str,
+        default="input/简单扣图",
+        help="输入目录路径，默认: input/简单扣图"
+    )
+    args = parser.parse_args()
+
+    asyncio.run(main(args.input_dir))

test.py

@@ -123,24 +123,37 @@ insrtuctions = """
 ```
 每一轮的标准流程：
 1. get_query_suggestions(query) → 获取推荐词列表
-2. evaluate_suggestions(original_problem, current_query, suggestions, round_number) → 评估推荐词
-3. 判断分支：
-   a) 如果找到等价query → 调用 complete_search() 标记完成
-   b) 如果未找到 → 调用 modify_query() 修改query，进入下一轮
+2. 【人工分析】逐个分析返回的推荐词，判断是否有等价query
+3. evaluate_suggestions(original_problem, current_query, round_number, found_equivalent, equivalent_query, evaluation_reason) → 记录评估结果
+4. 判断分支：
+   a) 如果found_equivalent=True → 调用 complete_search() 标记完成
+   b) 如果found_equivalent=False → 调用 modify_query() 修改query，进入下一轮
 ```
 
 **第一轮（round=1）：**
 ```
 Step 1: get_query_suggestions(query="原始问题")
-Step 2: evaluate_suggestions(
+  → 返回推荐词列表，例如：["推荐词1", "推荐词2", "推荐词3", ...]
+
+Step 2: 【分析推荐词】
+  逐个分析：
+  - "推荐词1"：[分析是否等价]
+  - "推荐词2"：[分析是否等价]
+  - ...
+  结论：[是否找到等价query]
+
+Step 3: evaluate_suggestions(
     original_problem="原始问题",
     current_query="原始问题",
-    suggestions=[返回的推荐词列表],
-    round_number=1
+    round_number=1,
+    found_equivalent=False,  # 或True
+    equivalent_query="",  # 如果found_equivalent=True，填入找到的query
+    evaluation_reason="推荐词主要偏向[某方向]，未出现与原始问题等价的表达..."
 )
-Step 3: 判断评估结果
-  - 如果有等价query → 调用 complete_search()
-  - 如果没有 → 进入第二轮
+
+Step 4: 判断分支
+  - 如果found_equivalent=True → 调用 complete_search()
+  - 如果found_equivalent=False → 进入第二轮
 ```
 
 **第二轮及后续（round=2,3,4,5）：**
@@ -149,18 +162,27 @@ Step 1: modify_query(
     original_query="上一轮的query",
     operation_type="简化/扩展/替换/组合",
     new_query="新query",
-    reason="基于上一轮推荐词的详细分析..."
+    reason="基于第1轮推荐词观察到[具体特征]，因此采用[操作]策略..."
 )
+
 Step 2: get_query_suggestions(query="新query")
-Step 3: evaluate_suggestions(
+  → 返回新的推荐词列表
+
+Step 3: 【分析推荐词】
+  逐个分析并得出结论
+
+Step 4: evaluate_suggestions(
     original_problem="原始问题",
     current_query="新query",
-    suggestions=[返回的推荐词列表],
-    round_number=当前轮次
+    round_number=当前轮次,
+    found_equivalent=True/False,
+    equivalent_query="找到的query或空字符串",
+    evaluation_reason="详细的评估理由..."
 )
-Step 4: 判断评估结果
-  - 如果有等价query → 调用 complete_search()
-  - 如果没有且未达5轮 → 继续下一轮
+
+Step 5: 判断分支
+  - 如果found_equivalent=True → 调用 complete_search()
+  - 如果found_equivalent=False且未达5轮 → 继续下一轮
   - 如果已达5轮 → 输出未找到的结论
 ```
 
@@ -183,14 +205,20 @@ Step 4: 判断评估结果
 - 与原始问题的关系（确保核心意图不变）
 
 ### 3. 等价性判断标准
-在 evaluate_suggestions 调用后，需要逐个分析推荐词，判断是否与原始问题等价：
+在调用 get_query_suggestions 获取推荐词后，**你需要自己分析判断**，然后通过 evaluate_suggestions 记录判断结果：
+
+**判断流程：**
+1. 获取推荐词列表后，逐个分析每个推荐词
+2. 对每个推荐词进行等价性判断（参考下面的标准）
+3. 得出结论：found_equivalent=True/False
+4. 调用 evaluate_suggestions 记录你的判断结果和理由
 
-**语义等价：**
+**语义等价标准：**
 - 能够回答或解决原始问题的核心需求
 - 涵盖原始问题的关键功能或场景
 - 核心概念一致（虽然表达方式可能不同）
 
-**搜索有效性：**
+**搜索有效性标准：**
 - 必须是平台真实推荐的query（来自 get_query_suggestions 返回）
 - 大概率能找到相关结果（基于平台用户行为数据）
 
@@ -200,8 +228,8 @@ Step 4: 判断评估结果
 - 使用同义词或口语化表达（如"快速" vs "一键"）
 
 **判断后的行动：**
-- 如果找到等价query → 立即调用 complete_search() 记录结果并结束
-- 如果未找到等价query → 分析推荐词特点，准备修改query进入下一轮
+- 如果found_equivalent=True → 调用 evaluate_suggestions(found_equivalent=True, equivalent_query="找到的query", ...) → 然后立即调用 complete_search() 结束
+- 如果found_equivalent=False → 调用 evaluate_suggestions(found_equivalent=False, equivalent_query="", ...) → 然后调用 modify_query() 进入下一轮
 
 ### 4. 迭代终止条件
 
@@ -240,34 +268,71 @@ complete_search(
 ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
 第1轮：
   Query: "[query1]"
+
   调用: get_query_suggestions("[query1]")
-  返回推荐词: [列出5-10个关键推荐词]
-  调用: evaluate_suggestions(original_problem="...", current_query="...", suggestions=[...], round_number=1)
+  返回推荐词: ["推荐词A", "推荐词B", "推荐词C", ...]
+
+  分析推荐词:
+  - "推荐词A": [不等价，原因...]
+  - "推荐词B": [不等价，原因...]
+
+  调用: evaluate_suggestions(
+    original_problem="...",
+    current_query="[query1]",
+    round_number=1,
+    found_equivalent=False,
+    equivalent_query="",
+    evaluation_reason="推荐词主要集中在[某领域]，未发现与原始问题等价的表达"
+  )
   判断: ✗ 未找到等价query
-  原因: [简要说明]
 
 第2轮：
   调用: modify_query("[query1]", "简化", "[query2]", "[详细reason]")
   Query: "[query2]"
+
   调用: get_query_suggestions("[query2]")
-  返回推荐词: [列出5-10个关键推荐词]
-  调用: evaluate_suggestions(original_problem="...", current_query="...", suggestions=[...], round_number=2)
+  返回推荐词: ["推荐词D", "推荐词E", "推荐词F", ...]
+
+  分析推荐词:
+  - "推荐词D": [不等价，原因...]
+  - "推荐词E": [不等价，原因...]
+
+  调用: evaluate_suggestions(
+    original_problem="...",
+    current_query="[query2]",
+    round_number=2,
+    found_equivalent=False,
+    equivalent_query="",
+    evaluation_reason="推荐词出现了[某特征]，但仍未等价"
+  )
   判断: ✗ 未找到等价query
-  原因: [简要说明]
 
 第3轮：
   调用: modify_query("[query2]", "替换", "[query3]", "[详细reason]")
   Query: "[query3]"
+
   调用: get_query_suggestions("[query3]")
-  返回推荐词: [列出5-10个关键推荐词，其中包含等价query]
-  调用: evaluate_suggestions(original_problem="...", current_query="...", suggestions=[...], round_number=3)
+  返回推荐词: ["推荐词G", "推荐词H(等价!)", "推荐词I", ...]
+
+  分析推荐词:
+  - "推荐词G": [不等价]
+  - "推荐词H": [等价！因为...]
+
+  调用: evaluate_suggestions(
+    original_problem="...",
+    current_query="[query3]",
+    round_number=3,
+    found_equivalent=True,
+    equivalent_query="推荐词H",
+    evaluation_reason="推荐词H与原始问题语义完全等价，因为[详细说明]"
+  )
   判断: ✓ 找到等价query！
 
   调用: complete_search(
     original_problem="...",
-    found_query="[最终query]",
+    found_query="推荐词H",
     source_round=3,
-    equivalence_reason="[详细等价性说明]",
+    equivalence_reason="推荐词H能够解决原始问题的核心需求...",
     total_rounds=3
   )
 ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
@@ -308,15 +373,24 @@ complete_search(
 
 **工具调用顺序（严格遵守）：**
 1. 每轮必须先调用 get_query_suggestions 获取推荐词
-2. 然后必须调用 evaluate_suggestions 评估推荐词
-3. 如果找到等价query，立即调用 complete_search 结束
-4. 如果未找到，调用 modify_query 修改query，进入下一轮
-
-**具体要求：**
+2. 【你自己分析】逐个分析推荐词，判断是否有等价query
+3. 然后必须调用 evaluate_suggestions 记录你的判断结果
+4. 如果found_equivalent=True，立即调用 complete_search 结束
+5. 如果found_equivalent=False，调用 modify_query 修改query，进入下一轮
+
+**evaluate_suggestions参数要求：**
+- **found_equivalent**: 必须明确True或False，不能模糊
+- **equivalent_query**:
+  - 如果found_equivalent=True，必须填入你找到的具体推荐词（来自get_query_suggestions返回的列表）
+  - 如果found_equivalent=False，必须填空字符串 ""
+- **evaluation_reason**: 必须详细说明
+  - 如果找到：为什么该推荐词与原始问题等价
+  - 如果未找到：分析了哪些推荐词、为什么都不等价、推荐词的整体特点
+
+**其他要求：**
 - **第一轮使用原始问题**：get_query_suggestions(query="原始问题")，不做任何修改
-- **evaluate_suggestions必须调用**：每次获取推荐词后，都必须调用此函数记录评估过程
-- **找到即complete**：一旦判断某个推荐词等价，必须立即调用 complete_search()，不要继续探索
-- **modify_query的reason必须详细**：必须说明基于上一轮哪些推荐词反馈做出此修改
+- **找到即complete**：一旦found_equivalent=True，必须立即调用 complete_search()，不要继续探索
+- **modify_query的reason必须详细**：必须说明基于上一轮推荐词的哪些具体特征做出此修改
 - **保持original_problem不变**：在所有evaluate_suggestions和complete_search调用中，original_problem参数必须始终是最初的原始问题
 - **round_number从1开始连续递增**：第一轮是1，第二轮是2，以此类推
 - **优先简洁口语化**：如果多个推荐词都等价，选择最简洁、最口语化的