【Python+微信】【企业微信开发入坑指北】4. 企业微信接入GPT,只需一个URL,自动获取文章总结

简介: 【Python+微信】【企业微信开发入坑指北】4. 企业微信接入GPT,只需一个URL,自动获取文章总结
  • 大家好,我是同学小张,日常分享AI知识和实战案例
  • 欢迎 点赞 + 关注 👏,持续学习持续干货输出
  • 一起交流💬,一起进步💪。
  • 微信公众号也可搜【同学小张】 🙏

本站文章一览:


经过这么多天的探索,终于到了激动人心的时刻。我们终于可以在企业微信和AI大模型应用方面做点稍微复杂点的功能了。

本文实现的效果如下:在微信中给企业微信发一个URL,它就能自动帮你总结出这个URL背后的网页所讲的内容。

想象一下,你平时在上网的过程中,看到一篇貌似很符合你需求的文章,但是并不能确定是否真的值得读。这时候,丢一个URL给微信,它给你输出总结,通过总结先来概览以下文章内容,再决定是不是要细看。是不是很有意义的一个功能?

其实市面上已经有很多类似的功能,比如Kimi Chat等,都有类似功能。本文是此类功能的一种实现方式,并且与微信联系了起来。

话不多说,开整。

1. 接收用户消息并处理

详细教程链接(如何接收用户消息,如何给用户回复消息等):【Python+微信】【企业微信开发入坑指北】1. 数据链路打通:接收用户消息处理并回复

1.1 主要代码

在之前代码的基础上,我们增加一下以下判断,只要发来的消息中带有 http 或 https 字样,就认为是一个url,就走我们本文的处理:爬取内容,总结内容。

if decrypt_data.get('Content', '').find("http://") != -1 or decrypt_data.get('Content', '').find("https://") != -1:
        response_content = await summary_url(decrypt_data.get('Content', ''))

当然,我也知道这种方式非常地… 不好,很容易误判,但对于目前的我来说够用了。实际工程中可能会考虑需要用户输入特定指令来触发该功能等更严格和标准的方式。

1.2 注意事项

前面我们也说了,企业微信服务只给我我们5s的响应时间,如果5s没返回响应,会触发重试。重试3次。

而对于本文所述功能来说,很容易超过5s。对于该问题的解决方法,前面也有文章讨论过

我这里也简单解决了一下该问题:用一个 Messages存储接收到的消息,当该消息已经在里面的时候,视为重试消息,直接返回一个 “success”。不在里面时,才视为新消息,触发后面的逻辑。这样保证了不会重复处理相同的请求。

Messages = {}
msg_id = decrypt_data.get('MsgId')
if (msg_id in Messages.keys()):
    return "success"
Messages[msg_id] = decrypt_data

然后,在消息处理完有结果之后,通过主动发送消息接口,将结果传给用户。

主动给企业微信发消息的教程参考:【Python+微信】【企业微信开发入坑指北】2. 如何利用企业微信API主动给用户发应用消息

if decrypt_data.get('Content', '').find("http://") != -1 or decrypt_data.get('Content', '').find("https://") != -1:
    response_content = await summary_url(decrypt_data.get('Content', ''))
    weichat_app.send_text(content=response_content)

2. 后台服务处理程序(通用爬虫+总结)

有了上面的与用户交互的通道之后,剩下的就是其背后的主要处理过程了。

用户发一个URL过来,后台服务应该解析该URL,获取网页中的内容,然后进行总结。

2.1 通用爬虫

获取网页中的内容,就是前面我们探讨过的爬虫的内容(【Python实用技能】爬虫升级之路:从专用爬虫到用AI Agent实现通用网络爬虫(适合小白)),为了能满足用户输入的任意URL,我们需要使用进阶版的通用爬虫,也就是利用AI Agent实现通用爬虫。

我们将之前的代码提取出来,稍微修改下作为一个通用爬虫模块来使用。

2.1.1 大模型写爬虫代码

def get_outline(page):
    soup = _get_soup(page.html)
    outline = []
    def process_element(element, depth):
        name = element.name
        if not name:
            return
        if name in ["script", "style"]:
            return
        element_info = {"name": element.name, "depth": depth}
        if name in ["svg"]:
            element_info["text"] = None
            outline.append(element_info)
            return
        element_info["text"] = element.string
        # Check if the element has an "id" attribute
        if "id" in element.attrs:
            element_info["id"] = element["id"]
        if "class" in element.attrs:
            element_info["class"] = element["class"]
        outline.append(element_info)
        for child in element.children:
            process_element(child, depth + 1)
    try:
        for element in soup.body.children:
            process_element(element, 1)
    except:
        logger.error("get outline error")
        outline = []
    return outline
PROMPT_TEMPLATE = """Please complete the web page crawler parse function to achieve the User Requirement. The parse \
function should take a BeautifulSoup object as input, which corresponds to the HTML outline provided in the Context.
``python
from bs4 import BeautifulSoup
# only complete the parse function
def parse(soup: BeautifulSoup):
    ...
    # Return the object that the user wants to retrieve, don't use print
``
## User Requirement
{requirement}
## Context
The outline of html page to scrabe is show like below:
``tree
{outline}
``
"""
class WriteCrawlerCode(Action):
    async def run(self, url, requirement):
        codes = {}
        codes[url] = await self._write_code(url, requirement)
        if codes[url] == None:
            return None
        return "\n".join(f"# {url}\n{code}" for url, code in codes.items()) ## 返回固定格式的url + 相应爬虫代码
    async def _write_code(self, url, query):
        page = await WebBrowserEngine().run(url)
        outline = get_outline(page)
        if len(outline) == 0:
            return None
        outline = "\n".join(
            f"{' '*i['depth']}{'.'.join([i['name'], *i.get('class', [])])}: {i['text'] if i['text'] else ''}"
            for i in outline
        )
        code_rsp = await self._aask(PROMPT_TEMPLATE.format(outline=outline, requirement=query))
        code = CodeParser.parse_code(block="", text=code_rsp)
        return code

这部分代码与之前基本一致,主要改动是其中加了一点错误的处理机制。该部分的输入是URL和用户想从中提取的信息。

2.1.2 自动执行爬虫代码

输入一个URL与其对应的爬虫代码,自动执行程序获取相关网页内容。

class RunCrawlerCode(Action):
    async def run(self, url, codes):
        code, current = codes.rsplit(f"# {url}", maxsplit=1)
        name = uuid4().hex
        module = type(sys)(name)
        exec(current, module.__dict__)
        page = await WebBrowserEngine().run(url)
        data = getattr(module, "parse")(page.soup)
        print(data)
        return str(data)  # 以字符串形式返回

2.1.3 爬虫工程师角色定义

定义一个爬虫工程师的角色,将上面的两个步骤添加进来。

_act函数中,如果执行的是 WriteCrawlerCode 动作,则传入相应的URL和用户需求Requirement。并需要将结果写入memory中,以供后面的动作取用。

后面的动作通过 msg = self.rc.memory.get(k=1)[0] 获取memory的最新结果,然后运行。

from metagpt.roles import Role
class CrawlerEngineer(Role):
    ......
    def __init__(self, **kwargs) -> None:
        super().__init__(**kwargs)
        self.set_actions([WriteCrawlerCode, RunCrawlerCode])
        
    async def _think(self) -> None:
        ......
    async def _act(self) -> Message:
        """Perform an action as determined by the role.
        Returns:
            A message containing the result of the action.
        """
        todo = self.rc.todo
        if type(todo) is WriteCrawlerCode:
            resp = await todo.run(url=self.url, requirement=self.requirement)
            logger.info(resp)
            if (resp == None):
                return None
            self.rc.memory.add(Message(content=resp, role=self.profile))
            return resp
        msg = self.rc.memory.get(k=1)[0]
        resp = await todo.run(url=self.url, codes=msg.content) # 返回必须是字符串
        logger.info(resp)
        return Message(content=resp, role=self.profile) # resp必须是字符串,MetaGPT中限制的
    async def _react(self) -> Message:
        ......

2.1.4 将以上内容封装调用

class CrawlerDataProvider():
    def __init__(self) -> None:
        pass
    
    async def run(self, url, requirement="获取正文中的所有文字内容,如果正文有code,将code也作为文字内容"):
        msg = "start"
        role = CrawlerEngineer(url = url, requirement = requirement)
        logger.info(msg)
        result = await role.run(msg)
        logger.info("\n=========================================\n")
        logger.info(result)
        return result

封装之后,外部只需要引入这个CrawlerDataProvider类,实例化一个对象就可以直接使用了。使用示例如下:

if __name__ == "__main__":
    url="https://mp.weixin.qq.com/s/2m8MrsCxf5boiH4Dzpphrg"
    requirement="获取标题,正文中的所有文字内容,如果正文有code,将code也作为文字内容"
    data_provider = CrawlerDataProvider()
    asyncio.run(data_provider.run(url, requirement))

2.2 文本内容总结

有了网页内容之后,就可以利用大模型对其进行总结。这里就简单总结一下:

PROMPT_TEMPLATE = """简要总结下面文字的内容:
"{text}"
简要总结:"""
class SummaryArticle():
    def __init__(self) -> None:
        pass
    
    def run(self, text):
        prompt = PROMPT_TEMPLATE.format(text = text)
        response_content = openai_wrapper.get_chat_completion(prompt)
        print("response content: ", response_content)
        return response_content

2.3 组合使用示例

async def summary_url(url):
    text_provider = CrawlerDataProvider()
    text = await text_provider.run(url = url)
    summary = SummaryArticle()
    response_content = summary.run(text = text)
    return response_content
if __name__ == "__main__":
    response = asyncio.run(summary_url(url = "https://mp.weixin.qq.com/s/L_gHW-_TIipmcyDcdQpZRA"))
    print(response)

3. 踩坑

(1)自动运行爬虫程序时,可能遇到 BeautifulSoup is not defined 的错误。一般原因是运行环境缺少 bs4 包。执行以下命令安装即可:

pip install bs4

(2)因为我的服务配置特别特别低,运行速度非常慢,所以经常出现网页获取超时等获取不到内容的情况,非常不稳定… 在自己笔记本上没问题。这个暂时无解。

如果觉得本文对你有帮助,麻烦点个赞和关注呗 ~~~


  • 大家好,我是 同学小张,日常分享AI知识和实战案例
  • 欢迎 点赞 + 关注 👏,持续学习持续干货输出
  • 一起交流💬,一起进步💪。
  • 微信公众号也可搜【同学小张】 🙏

本站文章一览:

相关文章
|
12天前
|
JavaScript Java 测试技术
基于ssm+vue.js+uniapp小程序的停车场微信小程序附带文章和源代码部署视频讲解等
基于ssm+vue.js+uniapp小程序的停车场微信小程序附带文章和源代码部署视频讲解等
22 6
|
12天前
|
JavaScript Java 测试技术
基于ssm+vue.js+uniapp小程序的宠物医院微信小程序附带文章和源代码部署视频讲解等
基于ssm+vue.js+uniapp小程序的宠物医院微信小程序附带文章和源代码部署视频讲解等
21 3
|
12天前
|
缓存 监控 前端开发
python开发中的技术选型与性能优化
在Python项目(一个基于Django和React的电商平台)开发中,面临技术选型、性能优化、架构设计和成本节约等问题。选用Django后端框架和React前端,利用Redux管理状态。为优化性能,进行了数据库索引优化、使用Redis缓存、异步处理(Celery)。采用微服务、RESTful API和代码复用提升架构灵活性。通过开源软件、云服务和注重代码质量降低成本。同时,借助日志记录、版本控制和监控工具有效排查故障。最终实现了一个稳定、高效且可扩展的平台。
19 1
|
1天前
|
机器学习/深度学习 人工智能 自然语言处理
豆瓣评分9.5!清华大牛熬夜整理的Python深度学习教程开发下载!
深度学习目前已经成为了人工智能领域的突出话题。它在“计算机视觉和游戏(AlphaGo)等领域的突出表现而闻名。 今天给小伙伴们分享的这份手册,详尽介绍了用 Python 和 Keras进行深度学习的探索实践,涉及计算机视觉、自然语言处理、生成式模型等应用。
|
1天前
|
JSON 数据可视化 数据处理
Python基础第九篇(Python可视化的开发)
Python基础第九篇(Python可视化的开发)
|
2天前
|
人工智能 数据可视化 数据挖掘
10个提高Python开发效率的工具
10个提高Python开发效率的工具
|
2天前
|
存储 数据挖掘 索引
Python streamlit框架开发数据分析网站并免费部署
使用Python的Streamlit框架,开发了一个在线数据分析工具,替代Excel查看设备温度CSV数据。通过pandas读取数据,matplotlib绘制图表。程序处理CSV,提取所需列,计算最大最小平均值,用户可多选查看特定数据。[GitHub](https://github.com/yigedaigua/MGHB)上有完整代码,应用已部署至Streamlit Cloud。
|
6天前
|
IDE 数据可视化 Shell
在Python的开发过程中,有许多工具可以帮助提高开发效率和代码质量
【6月更文挑战第14天】在Python的开发过程中,有许多工具可以帮助提高开发效率和代码质量
25 2
|
8天前
|
Python
别再被Python的面向对象编程绕晕了!一篇文章带你理清思路!
【6月更文挑战第12天】了解Python的面向对象编程(OOP)至关重要,它基于“万物皆对象”的理念,通过类(Class)定义对象的属性和方法。对象是类的实例,具有状态和行为。OOP包括类、对象、属性和方法四个基本概念。例如,`class Dog`定义了一个有`name`和`age`属性及`bark`方法的类。OOP提供代码重用、模块化、可扩展性和易理解性等优势,是现代编程的常用范式。通过学习,你可以更好地掌握这一核心技能。
|
8天前
|
存储 监控 Java
python自研流星监控系统meteor_monitor(第一篇)
本文介绍了作者开发的一个Python流星监控系统,替代了性能不佳且收费的ufocapturehd2软件。系统采用Win10相机应用低耗录制视频,通过SikuliX进行自动化控制,分段录制并存储到本地,然后通过脚本同步到NAS。视频分析使用帧差法检测流星,支持分布式分析。代码已更新,旧文章不再适用,最新内容可见:[用python自行开发的流星监控系统meteor_monitor(第二篇)-CSDN博客](https://github.com/xingxinghuo1000/meteor_monitor_scripts.git)。

热门文章

最新文章