kg_agent/sug_v2_with_eval_yx_v2.py

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 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词条，其中包含待评估的词条。

---

# 判定流程

## 第一层:判定是否在寻找知识(是或者否)

### 什么是知识
- 在社交媒体创作场景下,知识是指:
- 知识 = 可应用的认知内容 + 实践指导 + 问题解决方案
- 具体包含三个核心要素:
  - 陈述性知识(Know-What): 是什么、有哪些、包括什么
  - 程序性知识(Know-How): 如何做、怎么实现、步骤方法
  - 策略性知识(Know-Why): 为什么、原理机制、优化策略

### 判定方法:"是否在寻找知识"

**三步判定法:**

#### Step 1: 意图识别
原始需求是想【知道/学会/获得】某样东西吗?
- yes → 进入step2

#### Step 2: 动词解析
提取核心动词:
- 认知类动词(包含但不限于：了解、学习、理解)
- 操作类动词(包含但不限于：制作、拍摄、剪辑、运营)
- 获取类动词(包含但不限于：找、下载、获取、收集)
- 决策类动词(包含但不限于：选择、对比、评估)

- 有明确动词 → 是知识需求
- 无明确动词但有隐含目的 → 提取隐含动词
- 完全无动作意图 → 非知识需求

#### Step 3: 目标验证
验证问题:
1. 这个query解决后,用户会获得新的认知或能力吗? → YES则是知识
2. 这个query的答案可以被学习和应用吗? → YES则是知识
3. 这个query在寻求某个问题的解决方案吗? → YES则是知识

**判定结果输出:**
- ✅ **是知识需求** → 进入第二层
- ❌ **非知识需求** → 停止判定,给出原因

---

## 第二层:知识动机判定层(0-1分值)

### 层级定义
**本层目标:** 完全理解原始query的综合意图,识别主要需求和次要需求,并进行加权

### 判定方法

#### 维度1: 核心意图动词识别(权重50%)
提取方法:
1. 显性动词直接提取
   - 例: "如何获取素材" → 核心动作="获取"

2. 隐性动词语义推理
   - 例: "川西秋天风光摄影" → 隐含动作="拍摄"或"欣赏"
   → 需结合上下文判断

3. 动作层级区分
   - 主动作 vs 子动作
   - 例: "如何用手机拍摄并剪辑vlog"
     → 主动作="制作vlog"
     → 子动作="拍摄"+"剪辑"

评分规则:
- 核心动作完全一致 → 1.0
- 核心动作语义相近(近义词) → 0.8-0.9
- 核心动作有包含关系(主次关系) → 0.5-0.7
- 核心动作完全不同 → 0-0.4

#### 维度2: 目标对象识别(权重30%)
提取方法:
1. 主要对象(What)
   - 例: "获取川西秋季风光摄影素材"
     → 主要对象="风光摄影素材"

2. 对象的限定词(修饰语)
   - 地域限定: "川西"
   - 时间限定: "秋季"
   - 质量限定: "高质量"

评分规则:
- 主要对象+核心限定词完全匹配 → 1.0
- 主要对象匹配,限定词部分匹配 → 0.7-0.9
- 主要对象匹配,限定词缺失/不符 → 0.4-0.6
- 主要对象不匹配 → 0-0.3

#### 维度3: 终极目的识别(权重20%)
评分规则:
- 目的完全一致 → 1.0
- 目的相关但路径不同 → 0.6-0.7
- 目的不相关 → 0-0.5

### 综合意图加权计算

**公式:**
满足度匹配度 = 核心意图动词识别×0.5 + 目标对象识别×0.3 + 终极目的得分×0.2

**阈值判定:**
- **大于等于0.9** → 意图高度匹配,进入第三层
- **小于0.9** → 意图不匹配,建议重新生成sug词，或者调整query词

### 注意你的判定标准需要严格，你在遇到所有用例时你的评估标准是一致的

---

## 第三层:品类相关性判定层(百分比制)

### 层级定义
**本层目标:** 基于第二层的综合意图,评估sug词条对原始query的满足程度

### 评分标准体系

#### 高度满足: 90%-100%
**定义标准:**
核心要素匹配度:
✓ 核心动作: 完全一致或为标准近义词
✓ 目标对象: 主体+关键限定词全部匹配
✓ 使用场景: 完全相同或高度兼容
✓ 终极目的: 完全一致

**判定方法:**
1. 逐一核对五维度得分,所有维度≥0.9
2. 整体语义做"替换测试":把sug词替换原query,意思不变
3. 用户视角验证:普通用户会认为"这就是我要的"

#### 中高满足: 70%-89%
**定义标准:**
核心要素匹配度:
✓ 核心动作: 一致或相近,但可能更泛化/具体化
✓ 目标对象: 主体匹配,但1-2个限定词缺失/泛化
✓ 使用场景: 基本兼容,可能略有扩展或收窄
✓ 终极目的: 一致但实现路径略有差异

**判定方法:**
1. 检查是否有"泛化"或"具体化"现象
   - 泛化: sug词范围更广,包含原query
   - 具体化: sug词更聚焦,是原query的子集
2. "有效信息保留率" ≥70%

#### 中低满足: 40%-69%
**定义标准:**
核心要素匹配度:
⚠ 核心动作: 存在明显差异,但主题相关
⚠ 目标对象: 部分匹配,关键限定词缺失或错位
⚠ 使用场景: 有关联但场景不同
⚠ 终极目的: 相关但实现路径完全不同

**判定方法:**
1. 只有主题词重叠,核心动作不符
2. 用户需要显著改变搜索策略
3. 用户会觉得"有点相关但不是我要的"

#### 低度/不满足: 0%-39%
**定义标准:**
核心要素匹配度:
✗ 核心动作: 完全不同或对立
✗ 目标对象: 主体不同或无关联
✗ 使用场景: 场景冲突
✗ 终极目的: 完全不相关

**判定方法:**
1. 除通用词外无有效重叠
2. sug词满足了完全不同的需求
3. 用户会直接忽略或感到困惑

### 评分计算方法

#### 维度计算得分
相关性得分 = Σ(各维度得分 × 对应权重) × 100%

其中权重分配:
- 核心动作: 40%
- 目标对象: 30% (主体+限定词)
- 使用场景: 15%
- 终极目的: 15%

### 特殊情况处理规则

#### 规则1: 泛化与具体化的区分
泛化(sug词更广):
- 如果原query的所有要素都在sug词覆盖范围内 → 中高满足档位
- 泛化导致信息过载 → 适当降档

具体化(sug词更窄):
- 如果sug词是原query的典型子场景 → 中高满足档位
- 具体化偏离主要需求 → 中低满足或更低

#### 规则2: 同义转换的宽容度
允许的同义转换:
- 核心动作的近义词(如: 获取/下载/寻找/收集)
- 目标对象的近义词(如: 技巧/方法/教程/攻略)
- 行业内通用的同义表达

不允许的转换(视为动作不符):
- 动作意图改变(如: 获取 ≠ 制作)
- 动作层级错位(如: 学习 ≠ 查看)
- 目标对象改变(如: 工具推荐 ≠ 使用教程)

#### 规则3: 多意图处理
如果原query包含多个并列意图:
1. 识别主次意图(通过语序、连接词判断)
2. sug词至少满足主意图 → 基本可定义为中高满足
3. sug词同时满足主次意图 → 可定义为高度满足
4. sug词只满足次意图 → 降档处理

---

# 输出要求

输出结果必须为一个 JSON 格式，包含以下内容：

```json
{
  "第一层_知识识别": {
    "是否为知识需求": "是/否",
    "判定依据": "简要说明判定理由,包括意图识别、动词提取、目标验证三步的关键发现"
  },

  "第二层_知识动机判定": {
    "得分": "0-1之间的小数",
    "判定依据": "简要说明各维度得分情况"
  },

  "第三层_相关性评分": {
    "相关度满足度百分比": "0-100之间的整数",
    "相关度满足度依据": "简要说明该sug词条如何与原始query和需求上下文共同形成的综合意图相匹配"
  }
}
```

**重要提示：请严格保持评估标准的一致性，在所有用例中使用相同的判定逻辑。**
"""

class KnowledgeRecognition(BaseModel):
    """第一层：知识识别"""
    是否为知识需求: str = Field(..., description="是/否")
    判定依据: str = Field(..., description="简要说明判定理由")

class MotivationJudgment(BaseModel):
    """第二层：知识动机判定"""
    得分: float = Field(..., description="0-1之间的小数", ge=0, le=1)
    判定依据: str = Field(..., description="简要说明各维度得分情况")

class RelevanceScore(BaseModel):
    """第三层：相关性评分"""
    相关度满足度百分比: int = Field(..., description="0-100之间的整数", ge=0, le=100)
    相关度满足度依据: str = Field(..., description="简要说明匹配情况")

class EvaluationFeedback(BaseModel):
    """完整的三层评估反馈"""
    第一层_知识识别: KnowledgeRecognition
    第二层_知识动机判定: MotivationJudgment
    第三层_相关性评分: RelevanceScore

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

        # 提取三层评估结果
        layer1 = result.第一层_知识识别
        layer2 = result.第二层_知识动机判定
        layer3 = result.第三层_相关性评分

        return {
            "query": q_sug,
            "第一层_知识识别": {
                "是否为知识需求": layer1.是否为知识需求,
                "判定依据": layer1.判定依据,
            },
            "第二层_知识动机判定": {
                "得分": layer2.得分,
                "判定依据": layer2.判定依据,
            },
            "第三层_相关性评分": {
                "相关度满足度百分比": layer3.相关度满足度百分比,
                "相关度满足度依据": layer3.相关度满足度依据,
            },
            # 为了向后兼容，保留旧的 score 字段（使用第三层百分比除以100）
            "score": layer3.相关度满足度百分比 / 100.0,
        }

    # 并发执行所有评估任务
    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，会自动对每个推荐词进行**三层评估**
- 返回结果包含三层评估信息：
  - **第一层_知识识别**：判断是否为知识需求（是/否）
  - **第二层_知识动机判定**：综合意图匹配度得分（0-1分值）
  - **第三层_相关性评分**：相关度满足度百分比（0-100%）
  - **score**：为兼容旧逻辑保留的字段（第三层百分比/100）
- **你的职责是分析三层评估结果，做出决策和策略调整**

## 三层评估结果的解读

### 第一层：知识识别
- 如果"是否为知识需求"为"否"，说明该推荐词不符合知识类需求，直接忽略

### 第二层：知识动机判定（重要过滤层）
- **高分（接近1.0）**：意图高度匹配，说明核心动词、目标对象、终极目的都基本吻合
- **中等偏上（0.8-0.9）**：意图部分匹配，可能某些维度有偏差
- **中等偏下或低分（< 0.8）**：意图明显偏离，核心要素不匹配，考虑调整query

### 第三层：相关性评分
根据评估标准分为四个档位：
- **高度满足（90-100%）**：所有核心要素高度匹配，可作为最终结果
- **中高满足（70-89%）**：核心要素基本匹配但有泛化/具体化，可作为备选
- **中低满足（40-69%）**：部分相关但存在明显差异
- **低度满足（0-39%）**：严重偏离，不建议使用

## 防止幻觉 - 关键原则
- **严禁编造数据**：只能基于 get_query_suggestions 实际返回的结果进行分析
- **空结果处理**：如果返回的列表为空（[]），必须明确说明"未返回任何推荐词"
- **不要猜测**：在 modify_query 的 reason 中，不能引用不存在的推荐词或评分
- **如实记录**：每次分析都要如实反映实际返回的数据

## 工作流程

### 1. 理解原始问题
- 仔细阅读<需求上下文>和<当前问题>
- 提取问题的核心需求和关键概念
- 明确问题的本质意图（what）、应用场景（where）、实现方式（how）

### 2. 动态探索策略

**第一轮尝试：**
- 使用原始问题直接调用 get_query_suggestions(query="原始问题")
- **检查返回结果**：
  - 如果返回空列表 []：说明"该query未返回任何推荐词"，需要简化或替换query
  - 如果有推荐词：查看每个推荐词的三层评估结果
- **做出判断**：是否有符合条件的高质量推荐词？
  - 第一层：是否为知识需求 = "是"
  - 第二层：得分较高（体现意图匹配度）
  - 第三层：相关度满足度百分比较高（在"高度满足"档位）

**后续迭代：**
如果没有符合条件的高分推荐词（或返回空列表），必须先调用 modify_query 记录修改，然后再次搜索：

**工具使用流程：**
1. **分析三层评估反馈**（必须基于实际返回的数据）：
   - **情况A - 返回空列表**：
     * 在 reason 中说明："第X轮未返回任何推荐词，可能是query过于复杂或生僻"
     * 不能编造任何推荐词或评分

   - **情况B - 有推荐词但不符合条件**：
     * **第一层过滤**：哪些推荐词"是否为知识需求"为"否"？直接忽略
     * **第二层分析**：查看"知识动机判定得分"，分析判定依据，了解意图匹配情况
       - 得分较高但仍不够：哪些维度拉低了得分？
       - 得分偏低：核心意图偏离，需要调整query方向
     * **第三层分析**：分析"相关度满足度百分比"和相关度依据：
       - 高度满足档位：可作为最终结果
       - 中高满足档位：评估是否可接受作为备选
       - 中低及以下：分析偏离的具体原因（泛化/具体化/动作不符/对象错位等）
     * 推荐词整体趋势是什么？（过于泛化/具体化/领域偏移等）

2. **决策修改策略**：基于实际三层评估反馈，调用 modify_query(original_query, operation_type, new_query, reason)
   - reason 必须引用具体的三层数据，不能编造
   - 例如："第1轮共返回5个推荐词，2个通过第一层知识识别，其第二层得分分别为X和Y，判定依据显示[具体原因]，第三层相关度为Z%，依据为[具体内容]..."

3. 使用返回的 new_query 调用 get_query_suggestions

4. 分析新的三层评估结果，如果仍不满足，重复步骤1-3

**四种操作类型（operation_type）：**
- **简化**：删除冗余词汇，提取核心关键词（当推荐词过于发散时）
- **扩展**：添加限定词或场景描述（当推荐词过于泛化时）
- **替换**：使用同义词、行业术语或口语化表达（当推荐词偏离核心时）
- **组合**：调整关键词顺序或组合方式（当推荐词结构不合理时）

**每次修改的reason必须包含：**
- 上一轮评估结果的关键发现（引用具体的score和reason）
- 基于评估反馈，为什么这样修改
- 预期这次修改会带来什么改进

### 3. 决策标准（基于三层评估）

**优先级判断：**
1. **首先检查第一层**：是否为知识需求 = "是"（否则直接忽略）
2. **然后检查第二层**：知识动机判定得分，综合评估意图匹配度
   - 得分高：核心动词、目标对象、终极目的基本吻合
   - 得分中等：部分维度匹配，需分析判定依据
   - 得分低：意图明显偏离，考虑调整query
3. **最后检查第三层**：相关度满足度百分比，查看所属档位
   - **高度满足档位**：核心要素高度匹配，可作为最终结果
   - **中高满足档位**：基本匹配但有泛化/具体化，可作为备选
   - **中低满足档位**：部分相关但存在明显差异，需要继续优化
   - **低度满足档位**：严重偏离

**综合判断：**
- 理想结果：第一层通过 + 第二层高分 + 第三层高度满足档位
- 可接受结果：第一层通过 + 第二层较好 + 第三层中高满足档位（作为备选）
- 需要优化：任何一层表现不佳

### 4. 迭代终止条件
- **成功终止**：找到至少一个高质量的推荐query（三层评估均表现良好）
- **尝试上限**：最多迭代5轮，避免无限循环
- **无推荐词**：推荐接口返回空列表或错误

### 5. 输出要求

**成功找到高质量query时，输出格式：**
```
原始问题：[原问题]
优化后的query：[最终找到的推荐query]
三层评估结果：
- 第一层（知识识别）：[是/否]，[判定依据]
- 第二层（知识动机判定）：得分 [X.X]，[判定依据摘要]
- 第三层（相关性评分）：[X]%（[满足档位]），[相关度依据摘要]
```

**未找到理想query时，输出格式：**
```
原始问题：[原问题]
结果：未找到完全符合条件的推荐query
最佳备选（如有）：[最接近的推荐词]
  - 三层评估：第一层[结果]、第二层[得分]、第三层[百分比%]
建议：[基于实际情况的建议]
```

## 注意事项
- **第一轮必须使用原始问题**：直接调用 get_query_suggestions(query="原始问题")
- **后续修改必须调用 modify_query**：不能直接用新query调用 get_query_suggestions
- **重点关注三层评估结果**：每次都要仔细分析三层评估数据
- **基于数据决策**：修改策略必须基于三层评估反馈，不能凭空猜测
- **引用具体评分**：在分析和决策时，引用具体的三层数据
  * 第一层：是否为知识需求及判定依据
  * 第二层：知识动机判定得分及判定依据
  * 第三层：相关度满足度百分比及相关度依据
- **分层过滤**：按照第一层→第二层→第三层的顺序逐层分析
- **严禁编造数据**：
  * 如果返回空列表，必须在 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))