lyg
2025-05-14 acde3bd32f07bf02839a21e8fe5b4e69bfca2251
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
# -*- coding: utf-8 -*-
# entity_recognition.py
# @author: lyg
# @date: 2025-04-24
# @version: 0.0.1
# @description: 实体抽取,将文本中的实体进行识别和提取。
 
from langchain_openai.chat_models import ChatOpenAI
from langchain_core.prompts import HumanMessagePromptTemplate, ChatPromptTemplate
from langchain_core.output_parsers import JsonOutputParser
import json
 
from knowledgebase import utils
from knowledgebase.doc.entity_helper import entity_helper
 
llm = ChatOpenAI(temperature=0,
                 model="qwen2.5-72b-instruct",
                 base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
                 api_key="sk-15ecf7e273ad4b729c7f7f42b542749e")
 
 
class EntityRecognition:
    """
    实体识别抽取。
 
    使用langchain构建实体抽取流程。
    """
    cache_file = "entity_recognition.cache"
 
    def __init__(self, doc_type: str):
        # 实体词列表
        entities = filter(lambda x: x.doc_type == doc_type, entity_helper.entities)
        entity_list = ';\n'.join([f'- {entity.name}:{entity.prompts}' for entity in entities]) + "。"
        msg = HumanMessagePromptTemplate.from_template(template="""
# 指令
请从给定的文本中提取实体词列表,实体词列表定义如下:
## 实体词列表及识别规则
""" + entity_list + """
# 约束
- 输出格式为JSON格式;
- 提取的实体词必须是上面列举的实体词;
- 输出数据结构为字符串数组。
# 示例
```json
["遥控帧格式","遥控包格式"]
```
 
# 文本如下:
{text}
"""
                                                       )
        prompt = ChatPromptTemplate.from_messages([msg])
        parser = JsonOutputParser(pydantic_object=list[str])
        self.chain = prompt | llm | parser
        self.cache = {}
        self.load_cache()
 
    def load_cache(self):
        """
        加载缓存。
        """
        if utils.file_exists(self.cache_file):
            text = utils.read_from_file(self.cache_file)
            self.cache = json.loads(text)
 
    def save_cache(self):
        """
        保存缓存。
        """
        text = json.dumps(self.cache)
        utils.save_to_file(text, self.cache_file)
 
    def run(self, in_text: str) -> list[str]:
        """
        运行实体识别抽取。
        :param in_text: str - 输入文本
        """
        # 缓存命中
        text_md5 = utils.generate_text_md5(in_text)
        if text_md5 in self.cache:
            return self.cache[text_md5]
        result = self.chain.invoke({"text": in_text})
        self.cache[text_md5] = result
        self.save_cache()
        return result