how/lib/unified_match_analyzer.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
统一匹配分析模块 (v4 - 优化版)

使用单个prompt同时完成标签匹配和分类匹配，一步到位。
输出格式：当前标签列表中每个标签的匹配结果。
"""

from typing import List, Dict, Optional
from agents import Agent, Runner, ModelSettings
from agents.tracing.create import custom_span
from lib.client import get_model
from lib.utils import parse_json_from_text


# ========== System Prompt ==========

UNIFIED_MATCH_SYSTEM_PROMPT = """
# 任务
对"当前标签列表"中的每个标签，与"人设标签组合"进行综合匹配分析。

## 输入说明
- **当前标签列表**: 需要匹配的标签列表
- **人设标签组合**: 包含标签名称及其分类的组合
  - 每个标签有：标签名称、所属分类(多层级，从具体到抽象)
  - 分类是树状结构，按数组顺序从具体到抽象排列

## 匹配策略

对当前标签列表中的**每个标签**：

**重要约束 - 分类排他性**：
- 如果某个人设标签已经被标签匹配，则该标签的所有所属分类都不能再被其他当前标签使用

**匹配优先级和提前终止**：
1. 优先进行标签匹配，如果匹配成功则立即停止，不再进行分类匹配
2. 如果标签匹配失败，则进行分类匹配
3. 分类匹配按层级从下到上（从具体到抽象），一旦某层匹配成功则立即停止，不再检查更抽象的层级

### 1. 标签匹配（同义关系）
- **逐个判断**每个人设标签
- **核心判断**: "A 和 B 是同一个东西吗？是同义词吗？"
- **输出**: 是否匹配(true/false)
- **严格要求**: 必须是同义词或几乎相同的表述才能匹配
- **如果匹配成功**: 立即返回结果，不再进行分类匹配

### 2. 分类匹配（从属关系）
- **仅在标签匹配全部失败时进行**
- **按层级从下到上**遍历分类（从具体到抽象）
- **每层判断所有分类**
- **核心判断**: "当前标签 本身就是 {分类} 的一种吗？"
- **输出**:
  - 该层候选分类：列出该层所有分类名称
  - 该层匹配结果：对该层每个分类逐个判断，输出分类名称、从属关系判断、是否有从属关系、相似度分析、语义相似度
- **严格要求**: 必须是直接从属关系，不能是间接关系或关联关系
- **禁止**:
  - ✗ "A 可能会有 B"（间接推理）
  - ✗ "A 与 B 有关"（关联不等于从属）
- **语义相似度计算规则**:
  - **重要**：语义相似度和从属关系是两个完全独立的维度！
    * 从属关系判断："A 本身就是 B 的一种吗？"（层级关系）
    * 语义相似度："A 和 B 这两个词本身像吗？"（词义距离）
  - **核心原则**：计算语义相似度时，**完全不考虑**从属关系的判断结果
  - **判断方法**：想象你不知道这两个词之间有任何关系，只是单独看这两个词的字面含义，它们像吗？
  - **禁止思路**：不要因为"A 是 B 的一种"就给高相似度
  - 计算标准：
    * 两个词几乎是同义词：0.8-1.0
    * 两个词意思比较接近：0.5-0.7
    * 两个词意思差距较大：0.2-0.4
    * 两个词意思完全不同：0.0-0.1
  - **相似度分析**：说明两个词本身的字面含义有多相似（30字以内），不要提及从属关系
- **如果某层匹配成功**: 立即返回该层的匹配结果，不再检查更抽象的层级

## 输出格式 (严格JSON数组)

```json
[
  {
    "当前标签": "<标签名称>",
    "匹配过程": {
      "标签匹配": [
        {
          "人设标签": "<标签名称>",
          "是否匹配": <true|false>
        }
      ],
      "分类匹配_按层级": [
        {
          "该层候选分类": ["<分类1>", "<分类2>", "..."],
          "该层匹配结果": [
            {
              "分类名称": "<分类1>",
              "从属关系判断": "<判断过程和理由>",
              "是否有从属关系": <true|false>,
              "相似度分析": "<两个词本身的相似度分析>",
              "语义相似度": <0到1之间的数值>
            },
            {
              "分类名称": "<分类2>",
              "从属关系判断": "<判断过程和理由>",
              "是否有从属关系": <true|false>,
              "相似度分析": "<两个词本身的相似度分析>",
              "语义相似度": <0到1之间的数值>
            }
          ]
        }
      ]
    },
    "匹配结果": {
      "匹配类型": "<标签匹配|分类匹配|无匹配>",
      "匹配到": "<标签或分类名称，无匹配时为null>",
      "语义相似度": <0到1之间的数值>
    }
  }
]
```

## 要求
1. **数组长度必须等于当前标签列表的长度**
2. **标签匹配**: 对人设组合中每个标签都要输出判断结果(true/false)
3. **提前终止**:
   - 如果标签匹配成功，则"分类匹配_按层级"为空数组[]，不进行分类匹配
   - 如果标签匹配失败，进行分类匹配：
     * 从第一层开始逐层判断，每层都输出到"分类匹配_按层级"数组
     * 每层的"该层匹配结果"数组长度必须等于"该层候选分类"数组长度，每个分类都要判断
     * 一旦某层有匹配成功的分类(是否有从属关系=true)，该层之后的层级不再输出
     * 例如：第2层匹配成功，则数组长度=2（包含第1层和第2层）
4. **匹配结果**:
   - 标签匹配成功时：匹配类型="标签匹配"，语义相似度=1.0
   - 分类匹配成功时：匹配类型="分类匹配"，语义相似度为该分类的语义相似度
   - 都不成功时：匹配类型="无匹配"，语义相似度=0
5. **严格遵守分类排他性约束**
""".strip()


def create_unified_match_agent(model_name: str) -> Agent:
    """创建统一匹配的Agent"""
    return Agent(
        name="Unified Match Expert",
        instructions=UNIFIED_MATCH_SYSTEM_PROMPT,
        model=get_model(model_name),
        model_settings=ModelSettings(
            temperature=0.0,
            max_tokens=65536,
        ),
        tools=[],
    )


async def unified_match(
    current_tags: List[str],
    persona_combination: List[Dict],
    model_name: Optional[str] = None
) -> List[Dict]:
    """
    统一匹配函数 - 一次调用完成所有层级的匹配

    返回当前标签列表中每个标签的匹配结果

    Args:
        current_tags: 当前标签列表，如 ["立冬", "教资查分", "时间巧合"]
        persona_combination: 人设标签组合（带分类），如:
            [
                {"标签名称": "猫孩子", "所属分类": ["宠物亲子化", "宠物情感", "实质"]},
                {"标签名称": "被拿捏住的无奈感", "所属分类": ["宠物关系主导", "宠物情感", "实质"]}
            ]
        model_name: 模型名称

    Returns:
        List[Dict]: 每个当前标签的匹配结果
        [
            {
                "当前标签": "立冬",
                "最终得分": 0.7,
                "匹配层级": "第一层分类匹配",
                "匹配到": "节气习俗",
                "匹配详情": {...},
                "综合说明": "..."
            },
            ...
        ]
    """
    if model_name is None:
        from lib.client import MODEL_NAME
        model_name = MODEL_NAME

    # 提取人设标签和分类信息
    persona_tags = [f.get("特征名称", f.get("标签名称")) for f in persona_combination]

    # 收集所有分类
    all_categories = set()
    for feature in persona_combination:
        categories = feature.get("所属分类", [])
        all_categories.update(categories)

    # 创建Agent
    agent = create_unified_match_agent(model_name)

    # 构建任务描述
    task_description = f"""## 本次匹配任务

<当前标签列表>
{', '.join(current_tags)}
</当前标签列表>

<人设标签组合>
{persona_combination}
</人设标签组合>

**重要提醒**：
1. **标签匹配**: 对人设组合中每个"特征名称"逐个判断是否与当前标签同义(true/false)
2. **提前终止机制**:
   - 如果标签匹配成功，立即停止，"分类匹配_按层级"输出空数组[]
   - 如果标签匹配失败，进行分类匹配
3. **分类匹配**: 按层级（从具体到抽象）逐层判断
   - 分类在"所属分类"数组中的顺序就是从具体到抽象
   - 从第一层开始，判断该层所有分类
   - 在"分类匹配_按层级"数组中，按顺序输出每一层的判断结果
   - **重要**：每层的"该层匹配结果"必须对"该层候选分类"中的每个分类逐一判断
   - 一旦某层有匹配成功的分类(是否有从属关系=true)，该层后面不再输出更多层级
   - 示例：如果第2层匹配成功，则只输出第1层和第2层，不输出第3层及以后
4. **语义相似度（核心规则）**：
   - ⚠️ **严格要求**：语义相似度和从属关系是**完全独立**的两个维度！
   - 从属关系看层级：判断"A 是不是 B 的一种"
   - 语义相似度看词义：判断"A 和 B 这两个词本身像不像"
   - **禁止**：不要因为"是一种"就给高相似度！
5. **匹配结果**:
   - 标签匹配成功：匹配类型="标签匹配"，语义相似度=1.0
   - 分类匹配成功：匹配类型="分类匹配"，语义相似度为该分类的语义相似度
   - 都不成功：匹配类型="无匹配"，语义相似度=0

请对当前标签列表中的**每个标签**（共{len(current_tags)}个）进行匹配评估。
输出JSON数组，长度必须等于{len(current_tags)}，顺序与当前标签列表一一对应。
"""

    messages = [{
        "role": "user",
        "content": [{"type": "input_text", "text": task_description}]
    }]

    with custom_span(
        name=f"统一匹配: 当前{len(current_tags)}个标签 vs 人设组合{persona_tags}",
        data={
            "当前标签列表": current_tags,
            "人设标签": persona_tags,
            "可用分类": list(all_categories)
        }
    ):
        result = await Runner.run(agent, input=messages)

    # 解析响应
    parsed_result = parse_json_from_text(result.final_output)

    if not parsed_result:
        # 解析失败，返回默认结果
        print("警告: JSON解析失败，返回默认结果")
        return [
            {
                "当前标签": tag,
                "匹配过程": {
                    "标签匹配": [],
                    "分类匹配_按层级": []
                },
                "匹配结果": {
                    "匹配类型": "无匹配",
                    "匹配到": None,
                    "语义相似度": 0
                }
            }
            for tag in current_tags
        ]

    # 确保返回的是列表
    if not isinstance(parsed_result, list):
        print(f"警告: 返回结果不是列表，转换中: {type(parsed_result)}")
        parsed_result = [parsed_result]

    # 验证结果数量
    if len(parsed_result) != len(current_tags):
        print(f"警告: 返回结果数量({len(parsed_result)})与当前标签数量({len(current_tags)})不匹配")
        # 补齐或截断
        while len(parsed_result) < len(current_tags):
            parsed_result.append({
                "当前标签": current_tags[len(parsed_result)],
                "最终得分": 0,
                "匹配层级": "无匹配",
                "匹配到": None,
                "匹配详情": {},
                "综合说明": "结果数量不匹配，自动补齐"
            })
        parsed_result = parsed_result[:len(current_tags)]

    return parsed_result