【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知识和实战案例
  • 欢迎 点赞 + 关注 👏,持续学习持续干货输出
  • 一起交流💬,一起进步💪。
  • 微信公众号也可搜【同学小张】 🙏

本站文章一览:

相关文章
|
1月前
|
Java
Java开发实现图片URL地址检验,如何编码?
【10月更文挑战第14天】Java开发实现图片URL地址检验,如何编码?
65 4
|
3月前
|
Python
揭秘Python编程核心:一篇文章带你深入掌握for循环与while循环的奥秘!
【8月更文挑战第21天】Python中的循环结构——for循环与while循环,是编程的基础。for循环擅长遍历序列或集合中的元素,如列表或字符串;而while循环则在未知循环次数时特别有用,基于某个条件持续执行。本文通过实例展示两种循环的应用场景,比如用for循环计算数字平方和用while循环计算阶乘。此外,还通过案例分析比较了两者在处理用户输入任务时的不同优势,强调了根据实际需求选择合适循环的重要性。
52 0
|
25天前
|
Web App开发 前端开发 程序员
将微信公众号文章同步到阿里云开发者社区
本文介绍了一种通过自己拓展的浏览器插件,便捷地将微信公众号文章同步到阿里云开发者社区的方法。
41 6
|
1月前
|
编解码 UED Python
Python批量修改指定目录下图片的大小名文章
Python批量修改指定目录下图片的大小名文章
16 1
|
2月前
|
存储 Go C语言
Python 的整数是怎么实现的?这篇文章告诉你答案
Python 的整数是怎么实现的?这篇文章告诉你答案
58 7
|
2月前
|
JSON API 数据库
从零到英雄?一篇文章带你搞定Python Web开发中的RESTful API实现!
在Python的Web开发领域中,RESTful API是核心技能之一。本教程将从零开始,通过实战案例教你如何使用Flask框架搭建RESTful API。首先确保已安装Python和Flask,接着通过创建一个简单的用户管理系统,逐步实现用户信息的增删改查(CRUD)操作。我们将定义路由并处理HTTP请求,最终构建出功能完整的Web服务。无论是初学者还是有经验的开发者,都能从中受益,迈出成为Web开发高手的重要一步。
59 4
|
1月前
|
存储 自然语言处理 数据库
Python字典操作实现文章敏感词检索
Python字典操作实现文章敏感词检索
|
1月前
|
存储 Java 开发者
用一篇文章告诉你如何篡改 Python 虚拟机
用一篇文章告诉你如何篡改 Python 虚拟机
13 0
|
2月前
|
小程序 API 开发工具
使用python 实现微信签到提醒功能
【9月更文挑战第4天】使用python 实现微信签到提醒功能
71 2
|
2月前
|
数据采集 存储 前端开发
Java爬虫开发:Jsoup库在图片URL提取中的实战应用
Java爬虫开发:Jsoup库在图片URL提取中的实战应用