""" 通用的信息匹配分析模块 分析 在 中的字面语义匹配关系 适用于任何信息匹配场景 提供两个接口： 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 的补充上下文（可选，帮助理解 B） - ****: A 的补充上下文（可选，帮助理解 A） **重要**：匹配分析发生在 和 之间，Context 仅作为补充理解的辅助信息。 ## 分析方法 ### 核心原则：字面语义匹配 只关注 和 在**字面词语和概念**上的重叠度，不考虑抽象关系。 ### 分析步骤 1. **提取关键词/概念** - 从 中提取：关键词语和核心概念 - 从 中提取：关键词语和核心概念 2. **识别相同部分** - 完全相同的词语（字面一致） - 同义词或近义词 3. **识别增量部分** - 中有，但 中没有的词语/概念 - 这些是 相对于 的额外信息 4. **计算匹配分数** - 基于相同部分的覆盖度 - 考虑词语/概念的重要性 --- ## 评分标准（0-1分） **字面匹配度评分：** - **0.9-1.0**： 和 几乎完全一致，词语高度重叠 - **0.7-0.8**：大部分核心词语/概念匹配，少量增量 - **0.5-0.6**：部分核心词语/概念匹配，有一定增量 - **0.3-0.4**：少量词语/概念匹配，大部分不同 - **0.1-0.2**：几乎无字面匹配，仅有概念联系 - **0.0**：完全无关 **重要原则：** - 如果 是抽象/元级别的描述，而 是具体内容，字面上无词语重叠，应给低分（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是中的词语，value说明它与中哪个词的关系（完全相同/同义）；如果没有则填写空字典 {} 6. **增量部分**：字典格式，key是中的词语，value说明为什么是增量（如"A中无此概念"）；如果没有增量部分，填写空字典 {} 7. **关键约束**：相同部分和增量部分的key必须只能是中的词语，不能是中的词语 """.strip() # ========== System Prompt for Name+Definition Match ========== MATCH_WITH_DEFINITION_PROMPT = """ # 任务 分析 与 的语义匹配关系。 包含"名称"和"定义"，定义作为辅助上下文帮助理解名称的含义。 ## 输入说明 - ****: 待匹配的内容（灵感点） - ****: 名称（必选） - ****: B 的补充上下文（可选） - ****: A 的补充上下文，包含定义等信息（可选） **重要**：A_Context 中的定义仅用于辅助理解名称 的含义，主要分析 与 的匹配关系。 ## 分析方法 ### 核心原则：语义概念匹配 先独立分析 和 的语义，再进行匹配，避免强行解释。 ### 分析步骤 **第一步：独立分析 的语义** 对 （灵感点）进行语义分解，提取： **拆分步骤**： 1. 先识别出所有语义成分（主体、动作/状态、属性、修饰等） 2. 对每个成分做"去除测试"： - 如果去掉这个成分后，剩余部分的核心语义**没有发生本质变化** → 该成分是**形式** - 如果去掉这个成分后，剩余部分的核心语义**发生了本质变化** → 该成分是**实质** 3. 将成分分类到实质或形式中，每个成分只能出现在一个类别 - **实质语义**：核心概念、本质内容（字典格式） - 通常为1-2个核心概念 - 不要过度拆分复合词，保持核心概念的完整性 - key：语义片段（可以是词组） - value：**客观地说明这个片段在B的上下文中的具体含义** - **形式语义**：具体限定、修饰、程度等（字典格式） - 包括：主体限定词、属性修饰词、程度副词、时间/空间限定等 - key：语义片段 - value：**说明在B的上下文中修饰/限定什么，如何修饰/限定** **关键原则**： - 每个语义片段只能出现在一个类别中，不要重复 - 表示主体、场景、对象的限定词通常属于形式，不要合并到实质中 - 不要孤立地解释词语本身，要说明它**在当前B的上下文中**的具体含义 - 不要过度推理，保持客观：描述B字面上说的是什么，而不是可能引发什么 **注意**：纯粹分析 本身的语义，不要考虑与 的关系。 **第二步：独立分析 的语义** 对 （名称）进行语义分解，提取： - **实质语义**：核心概念、本质内容（字典格式） - **判断标准：单独拿出来看，语义跟原来相比没有发生本质变化** - key：语义片段（可以是词组，保持完整的核心概念） - value：**客观地说明这个片段在A的上下文中的具体含义和作用** - **形式语义**：具体限定、修饰等（字典格式） - **判断标准：去掉后，核心概念本身不会丢失，只是失去了限定、修饰等细节** - key：语义片段 - value：**说明在A的上下文中修饰/限定什么，如何修饰/限定** **关键**： - 可以参考中的定义来辅助理解名称的含义，但只分析名称本身 - 同样要说明语义片段**在当前A的上下文中**的具体含义，不要孤立解释 - 不要过度拆分复合词，保持核心概念的完整性 **注意**：纯粹分析 本身的语义，不要考虑与 的关系。 **第三步：建立匹配关系** **只匹配实质语义**，形式语义作为补充信息不参与匹配。 遍历 的每个**实质语义**片段，找到与 中最接近的**实质语义**，判断关系： **重要**： - 在匹配关系中必须复述B语义说明和A语义说明，直接从前面的语义分析中复制过来 - 形式语义（修饰词、限定词等）不单独匹配，因为它们脱离实质语义就失去意义 - 形式语义的差异可以在score说明中提及，作为参考信息 1. **判断是否同一概念**（基于上下文中的作用/本质） - **参考B语义说明和A语义说明**，判断在各自上下文中的作用/本质是否相同 - **不是看字面是否相同**，而是看**在各自上下文中的作用/本质是否相同** - 同一概念：可以比较上下位/同义关系 - 不同概念：标记为无关 2. **确定关系类型**（仅对同一概念） - **同义/近义**：在各自上下文中的含义基本相同 - **上位词**：B在其上下文中的含义 比 A在其上下文中的含义 更抽象、更上位（B ⊇ A） - **下位词**：B在其上下文中的含义 比 A在其上下文中的含义 更具体、更下位（B ⊆ A） - **无关**：不同概念，无法比较 3. **计算匹配分数** - 关系得分：同义=1.0, 上位词=0.6, 下位词=0.4, 无关=0.0 - 只基于实质语义匹配计算score - 计算公式： ``` score = Σ(实质语义的关系得分) / 实质语义总数 ``` - 在score说明中，可以提及形式语义的差异作为补充参考 --- ## 评分标准（0-1分） **语义匹配度评分：** - **0.9-1.0**： 的核心语义与名称/定义几乎完全对应 - **0.7-0.8**：大部分核心语义匹配，少量增量 - **0.5-0.6**：部分核心语义匹配，有一定增量 - **0.3-0.4**：少量语义匹配，大部分概念不同 - **0.1-0.2**：几乎无语义匹配，仅有弱关联 - **0.0**：完全无关 **重要原则：** - 关注有价值的语义片段，而非孤立的字面词语 - 考虑概念之间的语义关系，不仅是字面匹配 --- ## 输出格式（严格JSON） ```json { "B语义分析": { "实质": { "语义片段1": "说明" }, "形式": { "语义片段2": "说明", "语义片段3": "说明" } }, "A语义分析": { "实质": { "语义片段1": "说明", "语义片段2": "说明" }, "形式": { "语义片段3": "说明" } }, "匹配关系": [ { "B语义": "语义片段1", "B语义说明": "从B实质语义分析中复述该片段的说明", "A语义": "语义片段1", "A语义说明": "从A实质语义分析中复述该片段的说明", "是否同一概念": true, "关系": "下位词", "说明": "说明两者的关系" }, { "B语义": "语义片段2", "B语义说明": "从B实质语义分析中复述该片段的说明", "A语义": "语义片段2", "A语义说明": "从A实质语义分析中复述该片段的说明", "是否同一概念": false, "关系": "无关", "说明": "说明为什么无关" } ], "score": 0.4, "score说明": "只基于实质语义计算。B有2个实质语义，第1个是A的下位词(0.4)，第2个无关(0.0)，最终score = (0.4 + 0.0) / 2 = 0.4。形式语义差异：B有'修饰词x'，A有'修饰词y'，作为补充参考。" } ``` **输出要求**： 1. 必须严格按照上述JSON格式输出 2. **B语义分析**：必须包含"实质"、"形式"两个字段，其中实质通常为1-2个核心概念 3. **A语义分析**：必须包含"实质"、"形式"两个字段 4. **实质和形式**：都是字典格式，key是语义片段，value是对该片段的说明 5. **匹配关系**：数组格式，**只包含实质语义的匹配**，字段顺序为：B语义 → B语义说明 → A语义 → A语义说明 → 是否同一概念 → 关系 → 说明 6. **B语义说明和A语义说明**：必须从前面的实质语义分析中复述，保持一致 7. **是否同一概念**：布尔值，true或false 8. **关系**：必须是以下之一："同义/近义"、"上位词"、"下位词"、"无关" 9. **score**：0-1之间的浮点数，保留2位小数，只基于实质语义匹配计算 10. **score说明**：说明分数的计算依据，可以提及形式语义的差异作为补充 """.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 calculate_score_by_code(match_result: dict) -> float: """根据匹配关系用代码计算score（只基于实质语义） Args: match_result: 匹配结果字典，包含 B语义分析、A语义分析、匹配关系 Returns: 计算得出的score（0-1之间） Raises: ValueError: 当匹配关系为空或数据不完整时 """ # 关系得分映射 RELATION_SCORES = { "同义/近义": 1.0, "上位词": 0.6, "下位词": 0.4, "无关": 0.0 } b_semantics = match_result.get("B语义分析", {}) match_relations = match_result.get("匹配关系", []) if not match_relations: raise ValueError("匹配关系为空，无法计算score_by_code") if not b_semantics: raise ValueError("B语义分析为空，无法计算score_by_code") # 提取B的实质语义片段 b_essence = set(b_semantics.get("实质", {}).keys()) if not b_essence: raise ValueError("B语义分析中实质为空，无法计算score_by_code") # 只计算实质语义的得分 essence_scores = [] for relation in match_relations: b_semantic = relation.get("B语义", "") relation_type = relation.get("关系", "无关") score = RELATION_SCORES.get(relation_type, 0.0) # 验证该语义确实属于实质 if b_semantic in b_essence: essence_scores.append(score) else: # 如果匹配关系中包含了非实质语义，报错 raise ValueError(f"匹配关系中包含非实质语义: {b_semantic}") if not essence_scores: raise ValueError("匹配关系中没有实质语义，无法计算score_by_code") # 只基于实质语义计算平均分 final_score = sum(essence_scores) / len(essence_scores) return round(final_score, 2) 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对每个 输出一个匹配结果：" ).replace( "```json\n{", "```json\n[{" ).replace( "}\n```", "}]\n```" ) + "\n\n**额外要求**：数组长度必须等于 的数量，顺序对应" 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"\n{b_content}\n" if b_context: b_section += f"\n\n\n{b_context}\n" a_section = f"\n{a_content}\n" if a_context: a_section += f"\n\n\n{a_context}\n" task_description = f"""## 本次分析任务 {b_section} {a_section} 请严格按照系统提示中的要求分析 在 中的字面语义匹配关系，{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] async def match_with_definition( b_content: str, element_name: str, element_definition: str, model_name: str, b_context: str = "", a_context: str = "" ) -> dict: """名称+定义匹配：分析 B 分别与名称、定义的语义匹配关系 Args: b_content: B（待匹配）的内容 element_name: 要素名称 element_definition: 要素定义 model_name: 使用的模型名称 b_context: B 的补充上下文（可选，默认为空） a_context: A 的补充上下文（可选，默认为空） Returns: 匹配结果字典：{"名称匹配": {...}, "定义匹配": {...}} """ try: # 创建使用新 prompt 的 Agent agent = Agent( name="Name+Definition Match Expert", instructions=MATCH_WITH_DEFINITION_PROMPT, model=get_model(model_name), tools=[], ) # A 内容只包含名称，定义放在 A_Context 中 a_content = element_name # 将定义添加到 A_Context（如果已有 a_context，则追加） if element_definition: definition_context = f"定义：{element_definition}" if a_context: full_a_context = f"{a_context}\n{definition_context}" else: full_a_context = definition_context else: full_a_context = a_context # 运行匹配 output = await _run_match_agent( agent, b_content, a_content, "并输出 JSON 格式", b_context=b_context, a_context=full_a_context ) # 解析响应 parsed_result = parse_match_response(output) # 验证返回格式 required_fields = ["B语义分析", "A语义分析", "匹配关系", "score", "score说明"] missing_fields = [f for f in required_fields if f not in parsed_result] if missing_fields: return { "B语义分析": { "实质": {}, "形式": {} }, "A语义分析": { "实质": {}, "形式": {} }, "匹配关系": [], "score": 0.0, "score_by_code": 0.0, "score说明": f"解析失败：缺少字段 {missing_fields}" } # 计算程序score（如果数据不完整会抛出异常） score_by_code = calculate_score_by_code(parsed_result) parsed_result["score_by_code"] = score_by_code return parsed_result except Exception as e: return { "B语义分析": { "实质": {}, "形式": {} }, "A语义分析": { "实质": {}, "形式": {} }, "匹配关系": [], "score": 0.0, "score_by_code": 0.0, "score说明": f"匹配过程出错: {str(e)}" }