github 地址

  • https://github.com/jiaqianjing/HowlerBot

代码

  • 文件 HowlerBot.tar.gz

B 站视频

主要代码展示

主要是展示主要代码逻辑和功能，不能直接运行哈！！！

定时推送

由于我们的整个项目是基于 python 协程实现，我们这里引入 AsyncIOScheduler 来协助我们完成一个定时的任务.

# 定时发送功能
async def food_delivery(bot: Optional[Wechaty]):
    rooms = await bot.Room.find_all()
    for sp_room in rooms:
        sp_room_topic = await sp_room.topic()
        log.info(f'sp room topic: [{sp_room_topic}]')
        ding = "干饭了！别忘用优惠券！打开淘宝，将口令粘贴到搜索框中。"
        await sp_room.say(ding)
        await sp_room.say(coupons)

global bot
bot = HowlerBot()
scheduler = AsyncIOScheduler()
trigger = CronTrigger(hour="10", minute="30")
scheduler.add_job(food_delivery,
                    trigger=trigger,
                    args=[bot],
                    coalesce=True,
                    misfire_grace_time=60)
scheduler.start()
await bot.start()

指令相关功能

  1. #干饭
  2. #唠嗑了
  3. #拜拜
  4. #你会啥
  5. #天气
async def on_message(self, msg: Message):
    from_contact = msg.talker()
    text = msg.text()
    send_contact = msg.to()
    room = msg.room()
    msg_type = msg.type()
    conversation: Union[Room,
                        Contact] = from_contact if room is None else room
    await conversation.ready()

    global chat_friend
    log.info(f"text: {text}")

    at_me = await msg.mention_self()

    rooms = await self.Room.find_all()
    log.info(f'rooms: {rooms}, nums: {len(rooms)}')

    # empty message (only open chat windows)
    if msg_type == MessageType.MESSAGE_TYPE_UNSPECIFIED:
        log.info(
            f"this msg may be empty. username: {conversation}, msg: {text}"
        )
        return

    if '#你会啥' in text:
        await conversation.say(help_doc)
        return

    if '#天气' in text:
        match_obj = re.match(r'^#天气_[u4e00-u9fa5]?', text)

        match_obj = re.match(r'^(#天气) ([\u4e00-\u9fa5]+)', text)

        if match_obj:
            log.info(f"match_obj.group(): {match_obj.group()}")
            weather = await get_weather(match_obj.group(2))
            await conversation.say(weather)
            return
        else:
            await conversation.say('请在#天气后空一格跟上要查寻的地址！例如: #天气 北京')
            return

    if '#干饭' == text:
        await conversation.say('打开淘宝，将口令粘贴到搜索框中:')
        await conversation.say(coupons)
        return

    if '#拜拜' == text:
        try:
            chat_friend.remove(conversation)
            ai_bot.history_clean()
        except Exception as e:
            log.error(e)
            return
        await conversation.say('下次唠！')
        return

    if '#唠嗑了' == text:
        chat_friend.append(conversation)
        await conversation.say('你想唠点啥？')
        return

    if conversation in chat_friend:
        data = ai_bot.response(text)
        await conversation.say(data)
        return

    if msg_type == Message.Type.MESSAGE_TYPE_IMAGE:
        #await conversation.say('已收到图像，文字提取中...')
        #file_box_user_image = await msg.to_file_box()
        #img_name = file_box_user_image.name
        #img_path = f'./recieve_img/{img_name}'
        #await file_box_user_image.to_file(file_path=img_path)
        #word = await get_content_from_img(img_path)
        #await conversation.say(word)
        return

    if msg_type == MessageType.MESSAGE_TYPE_RECALLED:
        recalled_msg = await msg.to_recalled()
        log.info(
            f"{from_contact.name} recalled msg: {recalled_msg.text()}")
        await conversation.say(f"{from_contact} recalled msg.")

# 查询天气的方法
async def get_weather(area: Optional[str]):
    params['area'] = area
    async with aiohttp.ClientSession() as session:
        async with session.get(url=url, headers=headers,
                               params=params) as resp:
            result = await resp.json()
            location = result['data']['location']
            print(f"result: {location}")
            zone = location['region'] if location['region'] else ''
            zone = f"{zone} {location['city']}" if location['city'] else zone
            zone = f"{zone} {location['district']}" if location[
                'district'] else zone
            realtime = result['data']['realtime']
            w = f"""{zone}
天气: {realtime['weather']},
时间: {realtime['time']},
温度: {realtime['airTempreture']},
风向: {realtime['windDirection']},
风力: {realtime['windForce']},
湿度: {realtime['humidity']},
空气质量: {realtime['index']['aqi']},
pm2.5: {realtime['index']['pm2.5']}
"""

            return w

#闲聊部分相关代码
def singleton(cls):
    instances = {}

    def wrapper(*args, **kwargs):
        if cls not in instances:
            instances[cls] = cls(*args, **kwargs)
        return instances[cls]

    return wrapper


@singleton
class DialogueBot:
    def __init__(self, model_name_or_path="plato-mini") -> None:
        self.seed = None
        self.min_dec_len = 1
        self.max_dec_len = 64
        self.num_return_sequences = 20
        self.decode_strategy = 'sampling'
        self.top_k = 5
        self.temperature = 1.0
        self.top_p = 1.0
        self.num_beams = 0
        self.length_penalty = 1.0
        self.early_stopping = False
        self.device = 'cpu'
        self.model = UnifiedTransformerLMHeadModel.from_pretrained(
            model_name_or_path)
        self.tokenizer = UnifiedTransformerTokenizer.from_pretrained(
            model_name_or_path)
        self.history = []

    def history_clean(self):
        del self.history[:]

    def response(self, input: Optional[str]) -> Optional[str]:
        print(self.history)
        self.history.append(input)
        inputs = self.tokenizer.dialogue_encode(
            self.history,
            add_start_token_as_response=True,
            return_tensors=True,
            is_split_into_words=False)
        inputs['input_ids'] = inputs['input_ids'].astype('int64')
        ids, scores = self.model.generate(
            input_ids=inputs['input_ids'],
            token_type_ids=inputs['token_type_ids'],
            position_ids=inputs['position_ids'],
            attention_mask=inputs['attention_mask'],
            max_length=self.max_dec_len,
            min_length=self.min_dec_len,
            decode_strategy=self.decode_strategy,
            temperature=self.temperature,
            top_k=self.top_k,
            top_p=self.top_p,
            num_beams=self.num_beams,
            length_penalty=self.length_penalty,
            early_stopping=self.early_stopping,
            num_return_sequences=self.num_return_sequences)
        bot_response = select_response(
            ids,
            scores,
            self.tokenizer,
            self.max_dec_len,
            self.num_return_sequences,
            keep_space=False)[0]
        self.history.append(bot_response)
        return bot_response

HowlerBot [外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-qZT8xqGc-1636677394608)(https://img.shields.io/badge/Wechaty-Contributor%20Program-green.svg)]

HowlerBot 定位是生活帮手，陪玩，整蛊等多元化的机器人。由于目前智商感人，有时候让人抓狂！不过她会成长，所以请给她点耐心，让我们见证脑残儿童的成长吧！

Prerequisites

  1. 需要获取 Wechaty Puppet Service 的 token
    1. 有关 token 的说明
    2. 申请免费试用地址
  2. python-wechaty 是 wechaty 的 python 客户端
    1. github 地址: https://github.com/wechaty/python-wechaty
  3. paddlepaddle, paddlenlp (用于一些本地模型推理，后期会分离出去，采用远端调用的方式请求，本地弊端太多)
    1. paddlepaddle github: https://github.com/PaddlePaddle/Paddle
    2. paddlenlp github: https://github.com/PaddlePaddle/paddlenlp

Note

  • 需要部署一个 wechaty puppet service 的 gateway 服务
    • 因为 wechaty 本身是 ts 实现的，为了支持多语言（例如本例的 python 调用）需要部署类似充当 “翻译” 的 proxy 服务，我们称为 gateway.
    • 具体步骤可以参见：https://python-wechaty.readthedocs.io/zh_CN/latest/introduction/use-padlocal-protocol/
    • 机器资源可以是你的本地电脑，也可以是云厂商的云机器，这里我是”薅“ 了百度云的轻量级服务器，新人活动 89 元/年，还是比较划算的。
    • 购置好机器后，安装 docker, 然后执行这个 start_puppet_service_gateway.sh (注意启动前，打开该文件，替换自己的 token)
  • 查询天气 功能这里调用是的百度云-云市场关于查询天气的第三方 API，在 header 中使用了鉴权的 appcode 这里脱敏了，如果需要此功能，需要用户自行解决。
  • 其他没啥注意的，按照 github 中的 readme 操作即可。

Usage

如何启动

  1. 需要修改 main.py 中环境变量 WECHATY_PUPPET_SERVICE_TOKEN 对应你的 puppet service gateway 的 token. 相关内容见 python-wechaty 文档

  2. 安装依赖

    pip install -r requirements

  3. 执行启动命令

    python main.py

目前机器人支持以下场景和指令

  1. #你会啥
    微信内推送帮助指令

  2. #唠嗑了
    开启闲聊模式，可以和你侃天侃地，你说一句，她回一句，不过不是简单一问一答的句式，而是她会在对话的过程中，主动询问，从而是整个聊天过程更加生动。从进入闲聊模式后，就记录了此次聊天的全部上下文在内存中（未来可以考虑引入缓存的管理算法，避免单次聊天内容过长，导致内存溢出），因此，在单轮闲聊模式中不会出现内容断层的现象。此外，她具备一些基础 knowledge，不会显得过于傻。通过 #拜拜 指令进行关闭闲聊模式，此时会释放此次对话的上下文内容。

  3. #拜拜
    关闭闲聊模式。（释放对话上下文信息）

  4. #天气 xxx
    查询天气预报，例如 #天气 北京。目前支持国内。

  5. #干饭
    会发出饿了吗外卖优惠券。除了使用该指令触发，还会定时每天 10 点半，主动提醒，让你错峰订餐，避免高峰等待过长时间用膳。

  6. 进群欢迎语
    当有新朋友进入机器人所在的群的时候，机器人会主动发送欢迎的消息。

Demo

一些具体的例子

  • 帮助指令
    [外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-v8rI8IKG-1636677394610)(./img/demo-05.png)]

  • 进群欢迎语
    [外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-xgogHqni-1636677394611)(./img/demo-01.png)]

  • 闲聊
    [外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-veQEnoT2-1636677394612)(./img/demo-02.png)]

  • 查询天气
    [外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-H2HitTRk-1636677394612)(./img/demo-03.png)]

  • 获取外卖优惠码
    [外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-Q2SjFbOr-1636677394613)(./img/demo-04.png)]

Citation

@misc{wechaty,
  author = {Huan LI, Rui LI},
  title = {Wechaty: Conversational RPA SDK for Chatbot Makers},
  year = {2016},
  publisher = {GitHub},
  journal = {GitHub Repository},
  howpublished = {\url{https://github.com/wechaty/wechaty}},
}
# paddlepaddle # nlp # 自然语言处理
Logo
百度飞桨AI Studio社区

学大模型，用大模型上飞桨星河社区！每天8点V100G算力免费领！免费领取ERNIE 4.0 100w Token >>>

更多推荐

cover

利用Amazon Bedrock畅玩Claude 3等多种领先模型，抢占AI高地(体验倒计时4小时)

avatar 百度飞桨星河社区
cover

LangChain的RAG实践

avatar 百度飞桨星河社区
cover

RAG综述汇总

avatar 百度飞桨星河社区