零风险!零付费!我把 AI 接入微信群,爸妈玩嗨了~附教程(下):大模型 API 接入
上篇,带大家玩转高德开放平台 API,为大模型提供和本地生活相关的可靠信息。
本文将带大家,结合免费的大模型API,基于微信机器人开发框架,打造完整的 Bot。
友情提醒:注册一个小号使用,严禁用于违法用途(如发送广告/群发/诈骗、色情、政治等内容),否则封号是早晚的事哦。
0. 开发框架选择
之前给大家分享过三种微信机器人的搭建方案,要么账号容易被封禁,要么需要付费使用。
有没有零风险,零付费的方案?
有的,本文将基于:搭建微信机器人的第4种方式,免费开源,轻量高效 中的框架进行实现,占用资源不多,而且可定制化程度非常高!
1. 大模型准备
尽管免费的 API 有速率限制,不过对于个人使用而言,完全足够!
前天,给大家盘点了免费且靠谱的大模型 API,统一封装,任性调用,赶紧用起来!
为了在应用中,任意切换各种大模型 API,我这里接入了所有兼容 OpenAI 格式的大模型,并进行统一封装,下面给出示例代码:
from openai import OpenAI
model_dict = {'gemini-1.5-pro': {'api_key': server_api_key,'base_url': server_url,'model_name': 'gemini-1.5-pro'},'ernie-128k': {'api_key': server_api_key,'base_url': server_url,'model_name': 'ERNIE-Speed-128K'},
}class UniLLM:def __init__(self):model_names = list(model_dict.keys())self.models = {name: LLM_API(api_key=model_dict[name]['api_key'], base_url=model_dict[name]['base_url'], model=model_dict[name]['model_name']) for name in model_names}def __call__(self, model_name, messages, temperature=0.7):model = self.models.get(model_name)return model(messages, temperature=temperature)
2. 微信机器人开发
参考:搭建微信机器人的第4种方式,相信看到这里的你,已经把服务部署好了。
首先,进行需求分析:一个完整的 Bot 应该具备哪些功能?
- 基本功能:
- 自动发送消息;
- 自动接收消息并处理。
- 特色功能:
- 定时任务;
- 本地生活服务。
为了快速实现 MVP 版本,这里优先完成以上功能。
来吧,一起一步步把 Bot 搭建好。
2.0 服务部署
参考上篇,采用 docker 一键部署,如果运行过程中遇到问题,可打印最后几行日志查看:
PS:如需更多定制化功能,也可选择源码部署,不过坑略多,下篇跟大家分享。
sudo docker logs --tail 40 wx
如果运行过程中新建了群,需要重启容器:
sudo docker restart wx # 容器名称 or 容器 id 均可
2.1 发送消息接口
这个比较简单,服务部署后就有了,再贴下代码:
def send_message(to='user', content='hello', isRoom=False):url = f'http://129.150.39.xxx:3001/webhook/msg/v2?token=123'data = {"to": to,"isRoom": isRoom,"data": {"content": content}}response = requests.post(url, headers={'Content-Type': 'application/json'}, data=json.dumps(data))if response.status_code == 200:print(response.json())else:print('发送接口调用失败')
2.2 接收消息接口
接收消息的接口,需要部署一个服务,这里我们选用 FastAPI 搭建。
接口定义如下:
app = FastAPI()@app.post("/receive")
async def receive_message(request: Request):# 处理请求数据data = await request.form()message_type = data.get("type")content = data.get("content")source = data.get("source")is_from_self = data.get("isMsgFromSelf")try:# 填写处理逻辑-开始handle_message(message_type, content, source, is_from_self)# 填写处理逻辑-结束return JSONResponse(content={"status": True, "data": {"type": "text", "content": "success"}})except Exception as e:return JSONResponse(content={"status": False, "data": {"type": "text", "content": "failed"}})if __name__ == "__main__":uvicorn.run(app, host="0.0.0.0", port=3000)
注意:上述接口需指定返回值,否则报错。
接下来,我们需要接收消息后的处理逻辑:
def handle_message(message_type='text', content='', source='', is_from_self="0"):# 初版只处理文本if message_type != "text":returnsource = json.loads(source)from_name = source['from']['payload']['name']if source['room']:if '@小爱' not in content: # 群聊中非@消息returncontent = content.replace("@小爱", "")# 处理逻辑-contentbot_answer = get_bot_answer(content)if source['room']:bot_answer = f'@{from_name} {bot_answer}'send_message(to=source['room']['payload']['topic'], content=bot_answer, isRoom=True)else:send_message(to=from_name, content=bot_answer, isRoom=False)return bot_answer
具体回复什么内容,封装在 get_bot_answer()
函数中,下面我们就来搞定它!
2.3 本地生活服务
上一篇中,我们已经搭建好了利用高德开放平台提供的本地生活服务API。
为了能够让 LLM 认识它们,并且合理调用他们,我拆解成了三个步骤:
- 首先,根据用户输入,判断用户意图;
- 然后,根据意图调用相应的API接口,获取可靠的本地生活信息;
- 最后,基于API返回的信息,LLM 整理后,回复用户。
我们一步步来搞定它!
2.3.1 意图识别
我这里预先定义了 6 种意图:['天气', '步行规划', '骑行规划', '驾车规划', '公交规划', '地点推荐']
,如果识别结果不在意图列表中,则直接调用 LLM 进行回答。
# 1. 首先判断意图
intention = intention_rec(user_content)
print(f'意图识别结果:{intention}')
if intention not in intentions_list:messages = [{'role': 'system', 'content': sys_base_prompt},{'role': 'user', 'content': user_content}]res = unillm(model_name=model_name, messages=messages)return res.strip() if res else f'小爱有点累了,稍候再试吧'
给大家展示下意图识别
的日志:
2.3.2 调用本地生活接口
接下来,根据意图识别
结果调用对应的本地生活接口,下面以天气预报
为例:
首先需要LLM 提取出用户输入
中的地址信息;
然后请求天气预报接口get_weather
获取天气;
最后 LLM 进行整理,返回给用户。
# 2. 根据意图进行相应的操作
## 2.1 天气预报
if intention == '天气':# 提取地址messages = [{'role': 'user', 'content': f'{user_content},提取这句话中的地址信息,只需回答地址'},]res_address = unillm(model_name='glm4-9b', messages=messages)res_address = res_address.strip()wea_cast = get_weather(res_address, extensions='all')if not wea_cast:return '未找到该地址的天气信息,建议地址信息:省/市/区/'forecasts = wea_cast['forecasts'][0]['casts']today_cast = json.dumps(forecasts[0], ensure_ascii=False)future_cast = json.dumps(forecasts[1:], ensure_ascii=False)# 调用 LLMmessages = [{'role': 'system', 'content': sys_weather_report},{'role': 'user', 'content': f'地名:{res_address};今日天气:{today_cast};未来三天:{future_cast};用户问题:{user_content}'},]res_weather = unillm(model_name=wea_model_name, messages=messages)return res_weather if res_weather else f'小爱有点累了,稍候再试吧'
2.3.3 效果对比
比如我问他:周末不知道干点啥好
下面是没有意图识别
的结果:
收不住啊,胡诌了个鲤中步行街
给我,这是什么地方?
接下来是意图识别
的结果(地点推荐):基于高德的检索信息,LLM 给我答复如下。
2.4 定时任务实现
我们简单实现两个定时任务:
- 每天播报指定地区的天气信息;
- 根据成员的生日日期,自动发送生日祝福。
如果你的服务部署在海外,强烈建议为服务器设置时区:
# 查看当前时区
timedatectl
# 设置
sudo timedatectl set-timezone Asia/Shanghai
# 验证
timedatectl
执行后,就不用像上篇那样在代码中指定时区了。
2.4.1 每日天气播报
指定要发送的群聊名称,和要播报的地区:
def send_weather(to='机器人测试', addresses=['上海杨浦区',]):for add in addresses:content = weather_forecast(add)send_message(to=to, isRoom=True, content=content.strip())
2.4.2 定期生日祝福
首先定义一个字典,用于存放群成员的生日信息:
birthday_dict = {'爸爸': '07-25','大哥': '07-25','二哥': '07-25','妈妈': '07-25','大姐': '07-25',
}
考虑到国内大部分小伙伴,常用阴历生日,这里可以设置一下阳历/阴历转换:
from lunarcalendar import Converter, Solar, Lunar
# 阴历转换为阳历
def lunar_to_solar(date, leap_month=False):year, month, day = date.split('-')lunar_date = Lunar(int(year), int(month), int(day), leap_month)solar_date = Converter.Lunar2Solar(lunar_date)date = f'{solar_date.year}-{solar_date.month:02d}-{solar_date.day:02d}'return date
每天定时遍历,如果今天有某位群成员的生日,则调用 LLM 发送生日祝福:
def send_birthday_wish(to='机器人测试'):today = str(datetime.now().date())today_lunar = solar_to_lunar(today)today_lunar = '-'.join(today_lunar.split('-')[1:])if today_lunar in birthday_dict.values():for name, birthday in birthday_dict.items():if birthday == today_lunar:content = birthday_wish(name, solar=today, lunar=today_lunar)send_message(to=to, isRoom=True, content=content.strip())
最后,把所有定时任务放到后台去跑吧:
if __name__ == '__main__':# 每天发送天气预报group_name = 'xxx'schedule.every().day.at("06:30").do(lambda: send_weather(to=group_name))# 设置生日祝福任务schedule.every().day.at("06:00").do(lambda: send_birthday_wish(to=group_name))while True:schedule.run_pending()time.sleep(1)
给大家看下测试效果:
还不知道怎么给家人发生日祝福?问问小爱
吧~
写在最后
汇集各路大模型 API,以及高德提供的本地生活接口,终于把一个简单的微信机器人捏完了。
项目还在迭代中,大家有更好的想法,欢迎评论区交流。
如果本文对你有帮助,不妨点个免费的赞和收藏备用。
为了方便大家交流,新建了一个 AI 交流群
,欢迎感兴趣的小伙伴加入。
小爱
也在群里,想进群体验的朋友,公众号后台「联系我」即可,拉你进群。