how/lib/match_analyzer.py

"""
通用的信息匹配分析模块

分析 <B> 在 <A> 中的字面语义匹配关系
适用于任何信息匹配场景

提供两个接口：
1. match_single(b_content, a_content, model_name, b_context="", a_context="") - 单个匹配
2. match_batch(b_items, a_content, model_name, b_context="", a_context="") - 批量匹配

支持可选的 Context 参数：
- b_context: B 的补充上下文（帮助理解 B）
- a_context: A 的补充上下文（帮助理解 A）
- Context 默认为空，不提供时不会出现在 prompt 中
"""
import json
from typing import List
from agents import Agent, Runner
from agents.tracing.create import custom_span
from lib.client import get_model


# ========== System Prompt ==========
MATCH_SYSTEM_PROMPT = """
# 任务
分析 <B> 在 <A> 中的字面语义匹配关系。

## 输入说明

- **<B></B>**: 待匹配的内容（必选）
- **<A></A>**: 上下文内容（必选）
- **<B_Context></B_Context>**: B 的补充上下文（可选，帮助理解 B）
- **<A_Context></A_Context>**: A 的补充上下文（可选，帮助理解 A）

**重要**：匹配分析发生在 <B> 和 <A> 之间，Context 仅作为补充理解的辅助信息。

## 分析方法

### 核心原则：字面语义匹配
只关注 <B> 和 <A> 在**字面词语和概念**上的重叠度，不考虑抽象关系。

### 分析步骤

1. **提取关键词/概念**
   - 从 <B> 中提取：关键词语和核心概念
   - 从 <A> 中提取：关键词语和核心概念

2. **识别相同部分**
   - 完全相同的词语（字面一致）
   - 同义词或近义词

3. **识别增量部分**
   - <B> 中有，但 <A> 中没有的词语/概念
   - 这些是 <B> 相对于 <A> 的额外信息

4. **计算匹配分数**
   - 基于相同部分的覆盖度
   - 考虑词语/概念的重要性

---

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

**字面匹配度评分：**
- **0.9-1.0**：<B> 和 <A> 几乎完全一致，词语高度重叠
- **0.7-0.8**：大部分核心词语/概念匹配，少量增量
- **0.5-0.6**：部分核心词语/概念匹配，有一定增量
- **0.3-0.4**：少量词语/概念匹配，大部分不同
- **0.1-0.2**：几乎无字面匹配，仅有概念联系
- **0.0**：完全无关

**重要原则：**
- 如果 <A> 是抽象/元级别的描述，而 <B> 是具体内容，字面上无词语重叠，应给低分（0.1-0.3）
- 优先考虑具体词语的匹配，而非抽象概念的包含关系

---

## 输出格式（严格JSON）
```json
{
  "score": 0.75,
  "score说明": "简要说明分数是如何计算的，基于哪些词语/概念的匹配",
  "相同部分": {
    "B中的词1": "与A中的'某词'完全相同",
    "B中的词2": "与A中的'某词'同义"
  },
  "增量部分": {
    "B中的词3": "A中无此概念"
  }
}
```

**输出要求**：
1. 必须严格按照上述JSON格式输出（score 和 score说明在最前面）
2. 所有字段都必须填写
3. **score字段**：必须是0-1之间的浮点数，保留2位小数
4. **score说明**：必须简洁说明评分依据（基于相同部分的覆盖度）
5. **相同部分**：字典格式，key是<B>中的词语，value说明它与<A>中哪个词的关系（完全相同/同义）；如果没有则填写空字典 {}
6. **增量部分**：字典格式，key是<B>中的词语，value说明为什么是增量（如"A中无此概念"）；如果没有增量部分，填写空字典 {}
7. **关键约束**：相同部分和增量部分的key必须只能是<B>中的词语，不能是<A>中的词语
""".strip()


def create_match_agent(model_name: str) -> Agent:
    """创建信息匹配分析的 Agent

    Args:
        model_name: 模型名称

    Returns:
        Agent 实例
    """
    agent = Agent(
        name="Information Match Expert",
        instructions=MATCH_SYSTEM_PROMPT,
        model=get_model(model_name),
        tools=[],
    )

    return agent


def parse_match_response(response_content: str) -> dict:
    """解析匹配响应

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

    Returns:
        解析后的字典
    """
    try:
        # 如果响应包含在 markdown 代码块中，提取 JSON 部分
        if "```json" in response_content:
            json_start = response_content.index("```json") + 7
            json_end = response_content.index("```", json_start)
            json_text = response_content[json_start:json_end].strip()
        elif "```" in response_content:
            json_start = response_content.index("```") + 3
            json_end = response_content.index("```", json_start)
            json_text = response_content[json_start:json_end].strip()
        else:
            json_text = response_content.strip()

        return json.loads(json_text)
    except Exception as e:
        print(f"解析响应失败: {e}")
        return {
            "相同部分": {},
            "增量部分": {},
            "score": 0.0,
            "score说明": f"解析失败: {str(e)}"
        }


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

    Args:
        model_name: 模型名称

    Returns:
        Agent 实例
    """
    # 批量匹配的 System Prompt（在单个匹配基础上修改输出格式）
    batch_prompt = MATCH_SYSTEM_PROMPT.replace(
        "## 输出格式（严格JSON）",
        "## 输出格式（JSON数组）\n对每个 <B> 输出一个匹配结果："
    ).replace(
        "```json\n{",
        "```json\n[{"
    ).replace(
        "}\n```",
        "}]\n```"
    ) + "\n\n**额外要求**：数组长度必须等于 <B> 的数量，顺序对应"

    agent = Agent(
        name="Batch Information Match Expert",
        instructions=batch_prompt,
        model=get_model(model_name),
        tools=[],
    )

    return agent


async def _run_match_agent(
    agent: Agent,
    b_content: str,
    a_content: str,
    request_desc: str,
    b_context: str = "",
    a_context: str = ""
) -> str:
    """运行匹配 Agent 的公共逻辑

    Args:
        agent: Agent 实例
        b_content: B 的内容
        a_content: A 的内容
        request_desc: 请求描述（如"并输出 JSON 格式"或"并输出 JSON 数组格式"）
        b_context: B 的上下文（可选）
        a_context: A 的上下文（可选）

    Returns:
        Agent 的原始输出
    """
    # 构建任务描述
    b_section = f"<B>\n{b_content}\n</B>"
    if b_context:
        b_section += f"\n\n<B_Context>\n{b_context}\n</B_Context>"

    a_section = f"<A>\n{a_content}\n</A>"
    if a_context:
        a_section += f"\n\n<A_Context>\n{a_context}\n</A_Context>"

    task_description = f"""## 本次分析任务

{b_section}

{a_section}

请严格按照系统提示中的要求分析 <B> 在 <A> 中的字面语义匹配关系，{request_desc}的结果。"""

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

    # 使用 custom_span 追踪匹配过程
    # 截断显示内容，避免 span name 过长
    b_short = (b_content[:40] + "...") if len(b_content) > 40 else b_content
    a_short = (a_content[:40] + "...") if len(a_content) > 40 else a_content

    with custom_span(
        name=f"匹配分析: {b_short} in {a_short}",
        data={
            "B": b_content,
            "A": a_content,
            "B_Context": b_context if b_context else None,
            "A_Context": a_context if a_context else None,
            "模式": request_desc
        }
    ):
        # 运行 Agent
        result = await Runner.run(agent, input=messages)

    return result.final_output


async def match_single(
    b_content: str,
    a_content: str,
    model_name: str,
    b_context: str = "",
    a_context: str = ""
) -> dict:
    """单个匹配：分析一个 B 在 A 中的匹配

    Args:
        b_content: B（待匹配）的内容
        a_content: A（上下文）的内容
        model_name: 使用的模型名称
        b_context: B 的补充上下文（可选，默认为空）
        a_context: A 的补充上下文（可选，默认为空）

    Returns:
        匹配结果字典：{"相同部分": {}, "增量部分": {}, "score": 0.0, "score说明": ""}
    """
    try:
        # 创建 Agent
        agent = create_match_agent(model_name)

        # 运行匹配
        output = await _run_match_agent(
            agent, b_content, a_content, "并输出 JSON 格式",
            b_context=b_context, a_context=a_context
        )

        # 解析响应
        parsed_result = parse_match_response(output)

        return parsed_result

    except Exception as e:
        return {
            "相同部分": {},
            "增量部分": {},
            "score": 0.0,
            "score说明": f"匹配过程出错: {str(e)}"
        }


async def match_batch(
    b_items: List[str],
    a_content: str,
    model_name: str,
    b_context: str = "",
    a_context: str = ""
) -> List[dict]:
    """批量匹配：分析多个 B 在 A 中的匹配（一次调用）

    Args:
        b_items: B列表（多个待匹配项）
        a_content: A（上下文）的内容
        model_name: 使用的模型名称
        b_context: B 的补充上下文（可选，默认为空）
        a_context: A 的补充上下文（可选，默认为空）

    Returns:
        匹配结果列表：[{"相同部分": {}, "增量部分": {}, "score": 0.0, "score说明": ""}, ...]
    """
    try:
        # 创建批量匹配 Agent
        agent = _create_batch_agent(model_name)

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

        # 运行匹配
        output = await _run_match_agent(
            agent, b_list_str, a_content, "并输出 JSON 数组格式",
            b_context=b_context, a_context=a_context
        )

        # 解析响应（期望是数组）
        parsed_result = parse_match_response(output)

        # 如果返回的是数组，直接返回；如果是单个对象，包装成数组
        if isinstance(parsed_result, list):
            return parsed_result
        else:
            return [parsed_result]

    except Exception as e:
        # 返回错误信息（为每个 B 创建一个错误条目）
        return [{
            "相同部分": {},
            "增量部分": {},
            "score": 0.0,
            "score说明": f"匹配过程出错: {str(e)}"
        } for _ in b_items]