使用RPA通过GPT大模型AI Agent自动执行业务流程任务企业级应用开发实战：企业级应用开发的挑战与解决方案

    import jieba
    from gpt_gen import GPTGenerator
    from typing import List, Tuple
    from collections import defaultdict
    import json
    
    def load_data() -> dict:
    """
    从文件中读取数据，返回字典类型数据
    :return: 
    """
    data = {}
    with open('data.json', 'r') as f:
        for line in f:
            q, a = json.loads(line)
            data[q] = a
    return data
    
    class AutoReplyModel():
    
    def __init__(self):
        self.generator = GPTGenerator('pretrain_model/') # GPT模型加载
        self.keyphrase_dict = self._load_keywords('keywords.txt') # 关键词加载
    
    def _load_keywords(self, file_path: str) -> dict:
        """
        加载关键词文件
        :param file_path: 
        :return: 
        """
        keyphrases = {}
        with open(file_path, 'r') as f:
            lines = [l.strip().split('\t')[0] for l in f if not l.startswith('#')]
            for k in lines:
                words = list(jieba.cut(k))
                keyphrases[''.join(words)] = len(words)
        return keyphrases
    
    def extract_keyphrases(self, sentence: str) -> List[str]:
        """
        提取关键词
        :param sentence: 用户输入句子
        :return: 返回句子中所有关键词
        """
        keywords = []
        sentences = list(jieba.cut(sentence))
        for i in range(len(sentences)):
            w = ''.join([w for w in sentences[:i+1]])
            if w in self.keyphrase_dict and len(w)>1:
                keyword = tuple([''.join(list(jieba.cut(word))) for word in jieba.cut(sentence)[max(0,i-self.keyphrase_dict[w]):i]] + [''.join(list(jieba.cut(word))) for word in sentences[i:]])
                if all([tuple(jieba.cut(sent))[j] == word or j>=len(tuple(jieba.cut(sent)))-len(keyword)+1 for sent, j, word in zip(sentences, range(i), keyword)]) and set(keyword)<set(sentences):
                    keywords.append((''.join(list(jieba.cut(word))), max(0,i-self.keyphrase_dict[w]), i))
        return [(words, start, end) for words, start, end in sorted(keywords, key=lambda x:x[-1])]
    
    def predict(self, question: str) -> str:
        """
        预测答案
        :param question: 用户输入的问题
        :return: 答案文本
        """
        # 获取候选关键词列表
        keyphrases = self.extract_keyphrases(question)
    
        # 根据关键词，找到对应技能
        skill = ''
        for phrase in keyphrases:
            if phrase[0] in self.keyphrase_dict:
                skill += '-' + '-'.join(phrase[:-1]).replace('-',' ')
    
        answer = ''
        if skill!= '':
            answers = self.get_answers(skill)
    
            # 对答案进行排序和筛选
            answer = ''
            scores = []
            counts = defaultdict(int)
            for ans in answers:
                for kw in keyphrases:
                    if kw[0].lower() in ans.lower():
                        count = sum([ans.lower().count(word.lower()) for word in re.findall('[a-zA-Z]+', ans.lower()) if any(char.isdigit() for char in word)]) * sum([ans.lower().count(kw[0].lower())]) / len(re.findall('[a-zA-Z]+', ans.lower())) - min(len(ans)-sum([ans.lower().count(word.lower()) for word in re.findall('[a-zA-Z]+', ans.lower()) if any(char.isdigit() for char in word)]), len(ans)//2) * (1-abs((start - end)/(end-start))/10)*(-scores.index(len(ans))+1)/len(scores)**0.5*(counts[ans]/sum(counts.values()))**2
                        if ans not in self.history and ans!='':
                            count *= 2
                        counts[ans] += 1 
                        break
    
                scores.append(count)
    
            indices = sorted(range(len(scores)), key=lambda i: (-scores[i], i))[:min(len(scores), 3)]
    
            candidates = [ans for i, ans in enumerate(sorted(answers)) if i in indices][:3]
    
            # 如果多个候选答案都具有相同的得分，则随机返回其中一个
            random.shuffle(candidates)
            answer = '\n'.join(candidates)
    
        else:
            # 当无法识别技能时，生成一些随机文本
            while True:
                text = self.generator.generate(prompt='问题：'+question+'\n答案：', temperature=0.9, top_p=0.9)
                if text is None or ('是' in text and '不是' in text) or len(text)<20: 
                    continue
                answer = text
                break
        return answer
    
    def get_answers(self, skill: str) -> List[str]:
        """
        通过技能名称查找答案
        :param skill: 技能名称
        :return: 答案列表
        """
        # TODO：查询数据库或文件获取答案
        pass
    
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
    
    代码解读

    import pandas as pd
    from sklearn.preprocessing import StandardScaler, MinMaxScaler
    from sklearn.ensemble import RandomForestClassifier
    from joblib import dump, load
    
    class OrderSystemAutomator():
    
    def __init__(self):
        scaler_path = 'order_scaler.joblib'
        clf_path = 'order_clf.joblib'
        if os.path.exists(scaler_path) and os.path.exists(clf_path):
            self.scaler = load(scaler_path)
            self.clf = load(clf_path)
        else:
            df = pd.read_csv('orders.csv')
            self.X = df.drop('label', axis=1).fillna('')
            self.y = df['label']
            self.scaler = StandardScaler()
            self.X = self.scaler.fit_transform(self.X)
            self.clf = RandomForestClassifier(n_estimators=100, n_jobs=-1)
            self.clf.fit(self.X, self.y)
            dump(self.scaler, scaler_path)
            dump(self.clf, clf_path)
    
    def predict(self, orders: List[Tuple[str, int]]) -> Dict[str, float]:
        X = np.zeros((len(orders), 6))
        for i, order in enumerate(orders):
            feature = self.get_features(order)
            X[i,:] = feature
    
        y = self.clf.predict_proba(X)[:,1]
        result = {os[0]:float(o) for o, os in zip(y, orders)}
        return result
    
    def get_features(self, order: Tuple[str, int]) -> List[float]:
        """
        为每个订单获取特征
        :param order: 订单元祖，格式为("时间|数量|颜色|重量",数量)
        :return: 订单的特征列表
        """
        features = np.zeros(6)
        info = order[0].split('|')
        time = datetime.datetime.strptime(info[0], '%Y-%m-%d %H:%M:%S').timestamp()-datetime.datetime(year=1970, month=1, day=1).timestamp()
        features[:2] = [time*100, order[1]]
        colors = {'红色':'red', '绿色':'green', '蓝色':'blue'}
        weights = {'轻':'light', '中':'medium', '重':'heavy'}
        try:
            color = next(k for k, v in colors.items() if v==info[2])
            weight = next(k for k, v in weights.items() if v==info[3])
            features[[3,4,5]][[colors.keys().index(color), weights.keys().index(weight)]] = 1
        except StopIteration:
            print('颜色和重量信息不正确！')
        return list(features)
    
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
    
    代码解读

    import requests
    import xmltodict
    import logging
    import numpy as np
    import time
    import threading
    
    logging.basicConfig(level=logging.INFO, format='%(asctime)s %(filename)s[line:%(lineno)d] %(message)s', datefmt='%Y/%m/%d %H:%M:%S')
    logger = logging.getLogger(__name__)
    
    class FactoryLineMonitor():
    
    def __init__(self):
        self.url = "http://localhost:8080"
        self.deviceIds = ["DeviceId1", "DeviceId2"]
        self.sensorNames = {"Temperature": ("Sensor1", "0"), 
                            "Humidity": ("Sensor2", "0")}
        self.running = False
        self.lock = threading.Lock()
        self.stop_event = threading.Event()
        self.interval = 60
        self.data = {did:defaultdict(list) for did in self.deviceIds}
    
    def run(self):
        self.running = True
        t = threading.Thread(target=self._monitor)
        t.start()
    
    def stop(self):
        self.running = False
        self.stop_event.set()
    
    def _monitor(self):
        logger.info('Start monitoring...')
        while self.running:
            results = {}
            for deviceId in self.deviceIds:
                reqUrl = "{}?{}".format(self.url+'/rest/DataGetController/'+deviceId, 
                                        "&".join(["{}={}".format(v[0],v[1]) for k,v in self.sensorNames.items()]))
                res = requests.get(reqUrl)
                doc = xmltodict.parse(res.content)['root']['item'][0]['deviceDetail']
                deviceType = doc['@devicetype']
                sensorValues = doc['sensorValueList']['sensorValue']
                values = {k:(float(v) if v.isnumeric() else "") for sv in sensorValues if '@name' in sv and sv['@name']==sn for sn,sv in self.sensorNames.items()}
                results[deviceId] = {'type':deviceType, **values}
            timestamp = round(time.time()*1000)
            self.save_data(results, timestamp)
            time.sleep(self.interval)
    
    def save_data(self, results, timestamp):
        with self.lock:
            for did, d in results.items():
                for k, v in d.items():
                    if isinstance(v, float):
                        self.data[did][k+'_'+str(timestamp)].append(round(v, 2))
    
    def get_last_data(self, deviceTypes=[]):
        with self.lock:
            now = round(time.time()*1000)
            last_data = {}
            for did, dd in self.data.items():
                type_filter = deviceTypes if deviceTypes!=[] else ['ALL']
                for k, timestamps in dd.items():
                    category, metric = k.split('_')
                    if category in type_filter or 'ALL' in type_filter:
                        value = np.mean([ts[-1] for ts in timestamps if now-ts[-1]<60*1000*15])
                        if value!="":
                            last_data[(did,category,metric)] = value
    
            return last_data        
    
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
    
    代码解读