how/lib/batch_match_analyzer.py

"""
批量匹配分析模块

分析单个特征与多个特征之间的语义匹配度（批量版本）

提供接口：
analyze_batch_match(phrase_a, phrase_b_list, model_name) - 批量分析匹配度

返回格式：
[
    {
        "特征": "...",
        "分数": 0.85,
        "说明": "..."
    },
    ...
]
"""
from typing import List
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 ==========
BATCH_MATCH_SYSTEM_PROMPT = """
# 任务
分析单个特征 <A> 与多个特征 <B_List> 之间的语义匹配度。

## 输入说明
- **<A></A>**: 待分析的特征（必选）
- **<B_List></B_List>**: 多个特征列表（必选）

**重要**：
1. 必须在同一个评分标准下对所有 B 进行评分，确保分数可比
2. **优先识别并给出高分**给与 <A> 相似度最高的特征
3. 严格区分高相似度和低相似度，避免分数过于集中

---

## 评分标准（0-1分）

**核心原则**：从 <B_List> 中找出与 <A> 最相似的特征，给予最高分，其他按相似度递减。

- **0.9-1.0**：几乎完全相同（同义词、可互换）
- **0.7-0.9**：非常接近、高度相关（强关联、核心相关）
- **0.5-0.7**：有一定关联（中等相关、间接关联）
- **0.3-0.5**：关系较弱（弱相关、边缘关联）
- **0.0-0.3**：几乎无关或完全无关

**评分策略**：
- 优先识别与 <A> 最相似的特征，给 0.7+ 高分
- 对明显无关的特征，果断给 0.0-0.3 低分
- 合理使用中间分数段，避免过度集中
- 确保分数有梯度，体现明确的相似度差异

---

## 输出格式（严格JSON数组）

```json
[
  {
    "特征": "第一个B的特征",
    "分数": 0.85,
    "说明": "简要说明评分依据"
  },
  {
    "特征": "第二个B的特征",
    "分数": 0.45,
    "说明": "简要说明评分依据"
  }
]
```

**输出要求**：
1. 数组长度必须等于 <B_List> 的长度，顺序一一对应
2. 分数必须是0-1之间的浮点数，保留2位小数
3. 所有评分必须使用相同的标准，分数之间可比
4. **必须有明显的分数梯度**，最相似的给高分，不相关的给低分
""".strip()


def create_batch_match_agent(model_name: str) -> Agent:
    """创建批量匹配分析的 Agent

    Args:
        model_name: 模型名称

    Returns:
        Agent 实例
    """
    agent = Agent(
        name="Batch Match Expert",
        instructions=BATCH_MATCH_SYSTEM_PROMPT,
        model=get_model(model_name),
        model_settings=ModelSettings(
            temperature=0.0,
            max_tokens=65536,
        ),
        tools=[],
    )

    return agent


def clean_json_text(text: str) -> str:
    """清理JSON文本中的常见错误

    Args:
        text: 原始JSON文本

    Returns:
        清理后的JSON文本
    """
    import re

    # 1. 移除数组元素之间的异常字符（如 trib{）
    # 匹配模式：逗号后面跟着任意非空白字符，直到遇到正常的对象开始 {
    text = re.sub(r',\s*[a-zA-Z]+\s*\{', r',\n  {', text)

    # 2. 移除对象之间的异常字符
    text = re.sub(r'\}\s*[a-zA-Z]+\s*\{', r'},\n  {', text)

    return text


def parse_batch_match_response(response_content: str) -> List[dict]:
    """解析批量匹配响应

    Args:
        response_content: Agent 返回的响应内容

    Returns:
        解析后的字典列表
    """
    try:
        # 使用 parse_json_from_text 函数进行健壮的 JSON 解析
        result = parse_json_from_text(response_content)

        # 如果解析失败（返回空字典），尝试清理后再解析
        if not result:
            print(f"首次解析失败，尝试清理JSON文本后重新解析...")
            cleaned_text = clean_json_text(response_content)
            result = parse_json_from_text(cleaned_text)

            # 如果清理后仍然失败
            if not result:
                print(f"清理后仍解析失败: 无法从响应中提取有效JSON")
                return [{
                    "特征": "",
                    "分数": 0.0,
                    "说明": "解析失败: 无法从响应中提取有效JSON"
                }]

        # 确保返回的是列表
        if not isinstance(result, list):
            return [result]

        return result
    except Exception as e:
        print(f"解析响应失败: {e}")
        return [{
            "特征": "",
            "分数": 0.0,
            "说明": f"解析失败: {str(e)}"
        }]


async def analyze_batch_match(
    phrase_a: str,
    phrase_b_list: List[str],
    model_name: str = None
) -> List[dict]:
    """批量分析匹配度

    Args:
        phrase_a: 待分析的特征
        phrase_b_list: 多个特征列表
        model_name: 使用的模型名称（可选，默认使用 client.py 中的 MODEL_NAME）

    Returns:
        匹配结果列表：[{"特征": "...", "分数": 0.85, "说明": "..."}, ...]
    """
    try:
        # 如果未指定模型，使用默认模型
        if model_name is None:
            from lib.client import MODEL_NAME
            model_name = MODEL_NAME

        # 创建 Agent
        agent = create_batch_match_agent(model_name)

        # 构建 B 列表字符串
        b_list_str = "\n".join([f"- {b}" for b in phrase_b_list])

        # 构建任务描述
        task_description = f"""## 本次分析任务

<A>
{phrase_a}
</A>

<B_List>
{b_list_str}
</B_List>

请分析 <A> 与 <B_List> 中每个特征的匹配度，输出 JSON 数组格式的结果。
重要：必须使用一致的评分标准！"""

        # 构造消息
        messages = [{
            "role": "user",
            "content": [
                {
                    "type": "input_text",
                    "text": task_description
                }
            ]
        }]

        # 使用 custom_span 追踪分析过程
        # 截断显示内容，避免 span name 过长
        a_short = (phrase_a[:30] + "...") if len(phrase_a) > 30 else phrase_a

        with custom_span(
            name=f"批量匹配分析: {a_short} vs {len(phrase_b_list)}个特征",
            data={
                "phrase_a": phrase_a,
                "phrase_b_list": phrase_b_list,
                "b_count": len(phrase_b_list)
            }
        ):
            # 运行 Agent
            result = await Runner.run(agent, input=messages)

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

        # 验证返回的结果数量
        if len(parsed_result) != len(phrase_b_list):
            print(f"警告: 返回结果数量 ({len(parsed_result)}) 与输入数量 ({len(phrase_b_list)}) 不匹配")
            # 补齐或截断
            while len(parsed_result) < len(phrase_b_list):
                parsed_result.append({
                    "特征": phrase_b_list[len(parsed_result)],
                    "分数": 0.0,
                    "说明": "结果数量不匹配，自动补齐"
                })
            parsed_result = parsed_result[:len(phrase_b_list)]

        return parsed_result

    except Exception as e:
        # 返回错误信息（为每个 B 创建一个错误条目）
        return [{
            "特征": b,
            "分数": 0.0,
            "说明": f"分析过程出错: {str(e)}"
        } for b in phrase_b_list]