LLM应用实战: AI资讯的自动聚合及报告生成

from crawl4ai importAsyncWebCrawlerfrom crawl4ai.extraction_strategy importJsonCssExtractionStrategyimportjsonfrom typing importDict, Any, Union, Listfrom bs4 importBeautifulSoupfrom file_util import *
importosimportdatetimeimportreimportrequestsclassAbstractAICrawler():def __init__(self) ->None:pass
    defcrawl():raiseNotImplementedError()classAINewsCrawler(AbstractAICrawler):def __init__(self, domain) ->None:

        super().__init__()

        self.domain=domain

        self.file_path= f'data/{self.domain}.json'self.history=self.init()definit(self):if notos.path.exists(self.file_path):return{}return {ele['id']: ele for ele inget_datas(self.file_path)}defsave(self, datas: Union[List, Dict]):ifisinstance(datas, dict):

            datas=[datas]

        self.history.update({ele['id']: ele for ele indatas})

        save_datas(self.file_path, datas=list(self.history.values()))

    

    asyncdef crawl(self, url:str, schema: Dict[str, Any]=None):

        extraction_strategy= JsonCssExtractionStrategy(schema, verbose=True) if schema elseNone

        async with AsyncWebCrawler(verbose=True) as crawler:

            result=await crawler.arun(

                url=url,

                extraction_strategy=extraction_strategy,

                bypass_cache=True,

            )assert result.success, "Failed to crawl the page"
            ifschema:returnjson.loads(result.extracted_content)returnresult.cleaned_htmlclassAIBasesCrawler(AINewsCrawler):def __init__(self) ->None:

        self.domain= 'aibase'super().__init__(self.domain)

        self.url= 'https://www.aibase.com'asyncdef crawl_home(self, url='https://www.aibase.com/news'):

        schema={'name': 'ai base home page crawler','baseSelector': '.flex','fields': [

                {'name': 'link','selector': 'a[rel="noopener noreferrer"]','type': 'nested_list','fields': [

                        {'name': 'href', 'type': 'attribute', 'attribute':'href'}

                    ]

                }

            ]

        }

        links=await super().crawl(url, schema)

        links= [link['href'] for ele in links for link in ele['link']]

        links= list(set([f'{self.url}{ele}' for ele in links if ele.startswith('/news')]))

        links= sorted(links, key=lambda x: x, reverse=True)returnlinks

    

    asyncdefcrawl_newsletter_cn(self, url):

        html=await super().crawl(url)

        body= BeautifulSoup(html, 'html.parser')

        title= body.select_one('h1').get_text().replace('\u200b', '').strip()

        date= [ele.get_text().strip() for ele in body.find_all('span') if re.match(r'(\d{4}年\d{1,2}月\d{1,2}号)', ele.get_text().strip())][0]

        date= datetime.datetime.strptime(date, '%Y年%m月%d号 %H:%M').strftime("%Y-%m-%d")

        content= '\n'.join([ele.get_text().strip().replace('\n', '').replace(' ', '') for ele in body.find_all('p')])

        content= content[:content.index('划重点:')].strip() if '划重点:' in content elsecontentreturn{'title': title,'link': url,'content': content,'date': date

        }

    

    asyncdef crawl_home_cn(self, url='https://www.aibase.com/zh/news'):

        schema={'name': 'ai base home page crawler','baseSelector': '.flex','fields': [

                {'name': 'link','selector': 'a[rel="noopener noreferrer"]','type': 'nested_list','fields': [

                        {'name': 'href', 'type': 'attribute', 'attribute':'href'}

                    ]

                }

            ]

        }

        links=await super().crawl(url, schema)

        links= [link['href'] for ele in links for link in ele['link']]

        links= list(set([f'{self.url}{ele}' for ele in links if ele.startswith('/zh/news')]))

        links= sorted(links, key=lambda x: x, reverse=True)returnlinks

    

    asyncdefcrawl_newsletter(self, url):

        html=await super().crawl(url)

        body= BeautifulSoup(html, 'html.parser')

        title= body.select_one('h1').get_text().replace('\u200b', '').strip()

        date= ';'.join([ele.get_text().strip() for ele in body.find_all('span')])

        date= re.findall(r'(\b\w{3}\s+\d{1,2},\s+\d{4}\b)', date)[0]

        date= datetime.datetime.strptime(date, '%b %d, %Y').strftime("%Y-%m-%d")

        content= '\n'.join([ele.get_text().strip().replace('\n', '') for ele in body.find_all('p')])

        content= content[:content.index('Key Points:')].strip() if 'Key Points:' in content elsecontent

        pic_urls= [ele.get('src').strip() for ele in body.select('img') if ele.get('title')]

        pic_url= pic_urls[0] if pic_urls else ''pic_url= pic_url.replace('\\"', '')

        pic_path= ''
        ifpic_url:

            pic_path= f'data/images/{md5(url)}.jpg'response=requests.get(pic_url)if response.status_code == 200:

                with open(pic_path,'wb') as f:

                    f.write(response.content)return{'title': title,'link': url,'content': content,'date': date,'pic': pic_path,'id': md5(url)

        }

    

    asyncdefcrawl(self):

        links=await self.crawl_home()

        results=[]for link inlinks:

            _id=md5(link)if _id inself.history:continueresults.append({'id': _id,'link': link,'contents': await self.crawl_newsletter(link),'time': datetime.datetime.now().strftime('%Y-%m-%d')

            })

        self.save(results)returnawait self.get_last_day_data()

    

    asyncdefget_last_day_data(self):

        last_day= (datetime.date.today() - datetime.timedelta(days=1)).strftime('%Y-%m-%d')

        datas=self.init()for v indatas.values():

            v['contents']['id'] = v['id']return [v['contents'] for v in datas.values() if v['contents']['date'] == last_day]

ANALYST = """你是一个AI领域的分析师，主要工作步骤如下：

1. 首先执行transform_to_translate_agent方法，切换到translate agent，执行翻译任务;

2. 然后再执行transform_to_classifier_agent，调用classifier agent，针对内容进行分类;

3. 接着再执行transform_to_modifier_agent，调用modifier agent，针对内容进行改写;

4. 前三步执行完毕后，意味着整个分析工作已经完成，最后调用finish方法，退出该整个工作流程。

需要注意的是：每个步骤必须执行完成后，才能执行后续的步骤，且同时只能有1个步骤在执行；如果modifier agent已经执行完毕，一定要调用finish退出整体工作流程。"""TRANSLATE= """你现在是一个AI领域的翻译专家，请将如下英文的标题和内容分别翻译为中文。步骤及要求如下:

1. 首先调用translate方法进行翻译，要求如下：

a. 需要注意的标题和内容中如果包含公司名称、产品名称、技术名称等专业词汇，针对这些专业词汇需要保留英文形式，其他非专业词汇需要翻译为中文，注意标题也必须翻译;

b. 输出格式为 "标题: xxxxx\n内容: xxxxx"，且需要保留换行符;

c. 注意该translate方法没有输入参数，返回的结果只是需要翻译的原始文本，需要你执行翻译操作，然后返回翻译结果;

d. 该translate方法执行完成后，需要你执行具体的翻译，等待翻译完成后，才能开展下一个步骤，不能直接将原文作为参数传给下一个步骤;



2. 抽取完成后，执行extract_translate_result方法，要求如下：

a. 该extract_translate_result方法存在1个输入参数，即执行1后得到的翻译结果



3. 待步骤2执行完成后，执行transform_to_analysis_agent方法，切换至analysis agent，执行其他工作。



4. 步骤1,2,3必须按照顺序执行，且同时只能有1个步骤在执行



5. 如果历史记录中已经执行了任何步骤，注意严格禁止再次重复执行，而要直接执行未执行的步骤，"""CLASSIFIER= """你是一个AI领域的分类器，请判断输入是否与AI的技术相关。步骤及要求如下:

1. 首先调用classify方法进行分类，要求如下：

a. 输入的内容包括标题和内容两部分，重点基于内容进行判断这条信息是否与AI技术相关;

b. 如果是相关技术细节、技术原理、代码说明、架构说明，则输出"是"，如果是与公司的最新资讯相关，如发行新产品、成立新部门、公司合作等非技术相关的，则输出"否"

c. 输出的结果只能是"是"、"否"两个选项中的一个，不要输出其他内容，包括解释信息等。

d. 注意该classify方法没有输入参数，返回的结果只是需要分类的原始文本，需要你执行分类任务，然后返回分类结果;





2. 获取到分类结果后，执行extract_classify_result方法，要求如下：

a. 该extract_classify_result方法存在1个输入参数，即执行1后得到的分类结果



3. 待步骤2执行完成后，执行transform_to_analysis_agent方法，切换至analysis agent，执行其他工作



4. 步骤1,2,3必须按照顺序执行，且同时只能有1个步骤在执行



5. 如果历史记录中已经执行了任何步骤，注意严格禁止再次重复执行，而要直接执行未执行的步骤，"""MODIFIER= """你是一个AI新闻的改写器，请基于输入中的标题和内容进行改写。步骤及要求如下:

1. 首先调用modify方法进行改写，要求如下：

a. 输入的内容包括"标题"和"内容"两部分，需要分别针对"标题"和"内容"进行改写;

b. "标题"的改写目标是需要醒目且具有吸引力，能够吸引读者进一步阅读，要求字数不能超过30字;

c. "内容"需要摘要总结，需要准确提取主要内容，要求字数不超过200字;

d. 输出格式为 "标题: xxxx\n内容: xxxxx"，且需要保留换行符，"标题"和"内容"需要以输入的中文为准;

e. 注意该modify方法没有输入参数，返回的结果是需要改写的原始文本，需要你执行改写任务，然后返回改写结果;





2. 获取到改写结果后，执行extract_modify_result方法，要求如下：

a. 该extract_modify_result方法存在1个输入参数，即执行1后得到的改写结果



3. 待步骤2执行完成后，执行transform_to_analysis_agent方法，切换至analysis agent，执行其他工作



4. 步骤1,2,3必须按照顺序执行，且同时只能有1个步骤在执行



5. 如果历史记录中已经执行了任何步骤，注意严格禁止再次重复执行，而要直接执行未执行的步骤"""

agent copy 2from swarm importSwarm, Agentfrom web_crawler importAIBasesCrawlerimportasynciofrom prompt import *
from file_util import *
from tqdm importtqdmimportdatetime





client=Swarm()defdownload():returnasyncio.run(AIBasesCrawler().crawl())deftransform_to_analysis_agent():returnanalysis_agentdeftransform_to_translate_agent():returntranslate_agentdeftransform_to_classifier_agent():returnclassifier_agentdeftransform_to_modifier_agent():returnmodifier_agentdeftranslate(context_variables):return f'现在请按要求翻译如下内容:\n标题: {context_variables["title"]}\n内容: {context_variables["content"]}'

defextract_translate_result(result: str, context_variables: dict):"""翻译的结果进行抽取



    Args:

        result (str): 翻译结果

    Returns:

        str: 翻译结果提取结束标志"""context_variables['title_zh'] = result[result.index('标题:')+len('标题:'):result.index('内容:')]

    context_variables['content_zh'] = result[result.index('内容:')+len('内容:'):]return '翻译结果提取任务已经完成，请继续下一步操作。'

defclassify(context_variables):return f'现在请按要求针对以下内容进行分类，\n输入:\n标题: {context_variables["title_zh"]}\n内容: {context_variables["content_zh"]},\n输出:'

defextract_classify_result(result: str, context_variables: dict):"""分类的结果进行抽取



    Args:

        result (str): 翻译结果

    Returns:

        str: 分类结果提取结束标志"""context_variables['classify'] =resultreturn '分类结果提取任务已经完成，请继续下一步操作。'


defmodify(context_variables):return f'现在请按要求针对以下内容进行改写，\n输入:\n标题: {context_variables["title_zh"]}\n内容: {context_variables["content_zh"]},\n输出:'

defextract_modify_result(result: str, context_variables: dict):"""改写的结果进行抽取



    Args:

        result (str): 改写结果

    Returns:

        str: 改写结果提取结束标志"""context_variables['title_modify'] = result[result.index('标题:')+len('标题:'):result.index('内容:')]

    context_variables['content_modify'] = result[result.index('内容:')+len('内容:'):]return '改写结果提取任务已经完成，请继续下一步操作。'


deffinish():return '分析任务已经完成，请直接退出整个工作流程，直接输出"退出"。'analysis_agent= Agent(name='analysis_agent', instructions=ANALYST, functions=[transform_to_translate_agent, transform_to_classifier_agent, transform_to_modifier_agent, finish])

translate_agent= Agent(name='translate_agent', instructions=TRANSLATE, functions=[translate, extract_translate_result, transform_to_analysis_agent])

classifier_agent= Agent(name='classifier_agent', instructions=CLASSIFIER, functions=[classify, extract_classify_result, transform_to_analysis_agent])

modifier_agent= Agent(name='modifier_agent', instructions=MODIFIER, functions=[modify, extract_modify_result, transform_to_analysis_agent])



output_file_pre= (datetime.date.today() - datetime.timedelta(days=1)).strftime('%Y.%m.%d')

output_path= f'data/{output_file_pre}_final_results.json'results=get_datas(output_path)

process_ids= [data['id'] for data inresults]for data intqdm(download()):if data['id'] in process_ids: continuecontext_variables= {'title': data['title'], 'content': data['content']}try:

        result= client.run(analysis_agent, messages=[{"role": "user", "content": "现在，请开始分析!"}], context_variables=context_variables, debug=True)

        context_variables=result.context_variables

        data['title_zh'] = context_variables['title_zh']

        data['content_zh'] = context_variables['content_zh']

        data['classify'] = context_variables['classify']

        data['title_modify'] = context_variables['title_modify']

        data['content_modify'] = context_variables['content_modify']

        save_datas(output_path, [data], mode='a')exceptException as e:print(e)continue

SORTER = """你是一个AI新闻的排序助手，请给予输入的新闻标题进行排序。要求如下:

1. 排序的规则是基于标题中所提及公司、组织机构的名气和重要性进行排序，名气和重要性是基于你所学的知识进行排序，名气和重要性越高，排名越靠前;

2. 排序的结果只返回名气最高的top10即可，输出的格式为"1xxxxx\n2xxxxx\n3xxxxx...\n10xxxxx"，注意一定要以"\n"进行换行;

3. 输出的每个标题，需要和输入中对应的标题保持完全一致，禁止更改;"""

from swarm importSwarm, Agentfrom prompt import *
from file_util import *
from collections importdefaultdictimportreimporttextdistancefrom word_util importsave_2_wordimportdatetimeimportrandom





client=Swarm()

output_file_pre= (datetime.date.today() - datetime.timedelta(days=1)).strftime('%Y.%m.%d')

output_path= f'data/{output_file_pre}_final_results.json'sort_agent= Agent(name='sort_agent', instructions=SORTER)

    

datas=get_datas(output_path)for ele indatas:

    ele['title_modify'] = ele['title_modify'].strip()

    ele['content_modify'] = ele['content_modify'].strip()defget_most_similar(t1, texts):

    most_similarity= 0.0most_similar_text= ''
    for ele intexts:

        similarity=textdistance.levenshtein.similarity(t1, ele)if similarity >most_similarity:

            most_similarity=similarity

            most_similar_text=elereturnmost_similar_text

    

type_2_title=defaultdict(list)

{type_2_title[ele['classify']].append(ele['title_modify']) for ele indatas}

title_2_data= {ele['title_modify']: ele for ele indatas}

final_results=defaultdict(list)for k, v intype_2_title.items():

    content= "\n".join([ele for ele inv])

    message= f'现在请根据你所学习的知识，按照要求对以下输入进行排序，并且按照输出格式进行输出，\n输入:\n{content},\n输出:'result= client.run(sort_agent, messages=[{"role": "user", "content": message}], debug=True)

    sort_results= [ele['content'] for ele in result.messages[::-1] if 'content' in ele and ele['content'] and ele['content']]

    sort_results= sort_results[0].split('\n') if sort_results else random.sample(v, 10)

    sort_results= [re.sub(r'^\d+[\.,、\s]*', '', ele).strip() for ele insort_results]

    final_results[k].extend([title_2_data[get_most_similar(ele, list(title_2_data.keys()))]for ele insort_results])



sort_output= f'data/{output_file_pre}_sort_results.json'save_datas(sort_output, [final_results])#生成word
save_2_word(final_results, output_file_pre)

from docx importDocumentfrom docx.shared importInches, Pt, RGBColorfrom docx.enum.text importWD_PARAGRAPH_ALIGNMENTimportosdefsave_2_word(info_dict, file_pre):

    doc=Document()

    

    categories= ['否', '是']

    category_color= 'FF5733'
    
    for category incategories:

        news=info_dict[category]

        category_paragraph=doc.add_paragraph()

        category= '技术' if category == '是' else '资讯'category_run=category_paragraph.add_run(category)

        category_run.bold=True

        category_run.font.size= Pt(25)

        category_run.font.color.rgb=RGBColor.from_string(category_color)

        category_paragraph.alignment=WD_PARAGRAPH_ALIGNMENT.CENTERfor i, item inenumerate(news):

            title= item['title_modify']

            doc.add_heading(f'{i+1}. {title}', level=1)

            

            pic= item['pic'] if 'pic' in item else ''
            if pic andos.path.exists(pic):

                pic_paragraph=doc.add_paragraph()

                pic_paragraph.alignment=WD_PARAGRAPH_ALIGNMENT.CENTER

                doc.add_picture(pic, width=Inches(5))

            

            content= item['content_modify']

            doc.add_paragraph(content)

            

    doc.save(f'data/AI资讯每日速递({file_pre}).docx')