阅读量:0
说明
我之前实现了简单UI来跟OriginBot交互，可以参考这里：古月居 - ROS机器人知识分享社区 但是由于我不是专业的前端开发，写UI还是比较耗时的，所以最近想修改一下这部分。
还有一个原因是，自己开发前端，如果想实现远程交互（不在同一wifi下），就一定需要一个云服务器来中转一下，这个是比较麻烦的。
我期望的交互效果是可以双向发送文字、图片和语音，找了一圈发现钉钉的「单聊机器人」可以满足所有要求，所以这篇博客记录一下如何为OriginBot接入钉钉单聊机器人。
创建钉钉单聊机器人
创建单聊机器人的步骤其实不复杂，可以参考：单聊机器人概述 - 钉钉开放平台
唯一需要注意的是消息接收模式要选择Stream模式，不要选择http模式即可。
如何实现收发消息
在钉钉开放平台上创建好单聊机器人后，还需要有一个服务来收发消息。
在OriginBot上创建一个文件dingtalk_runtime.py, 内容如下：
""" 用于钉钉单聊机器人收发消息 """ from dingtalk_stream import AckMessage import dingtalk_stream import os from prompts import base_prompt from llms import azure_gpt4o class DingtalkMsgHandler(dingtalk_stream.ChatbotHandler): def __init__(self): super(dingtalk_stream.ChatbotHandler, self).__init__() async def process(self, callback: dingtalk_stream.CallbackMessage): incoming_message = dingtalk_stream.ChatbotMessage.from_dict(callback.data) message_type = incoming_message.message_type if message_type not in ("text"): self.reply_text( "您发送的消息类型不合法，目前只支持文本。", incoming_message ) return AckMessage.STATUS_OK, "OK" if message_type == "text": text = incoming_message.text.content.strip() self.reply_text(text, incoming_message) return AckMessage.STATUS_OK, "OK" def main(): credential = dingtalk_stream.Credential( os.getenv("DINGTALK_CLIENTID"), os.getenv("DINGTALK_CLIENTSECRET"), ) client = dingtalk_stream.DingTalkStreamClient(credential) client.register_callback_handler( dingtalk_stream.chatbot.ChatbotMessage.TOPIC, DingtalkMsgHandler() ) client.start_forever() if __name__ == "__main__": main() 运行这个脚本后，在钉钉中给“originbot_home_assistant”这个机器人发送消息后，它会给你回复一样的内容。
到这里就已经实现了最基础的交互功能了。
大家在代码中可以看到，我目前限定了只能接收文本格式的消息，其他类型暂时都不允许，这主要是为了降低一开始的开发难度，不用考虑所有可能。
但实际上，钉钉的单聊机器人支持很丰富的消息类型，可以看下面的说明：企业机器人发送单聊消息 - 钉钉开放平台
我会在后面需要的时候添加其他消息类型。
集成GPT4o
上面给出的通过钉钉单聊机器人跟智能小车交互的代码是非常简单的，它只能把你发给小车的消息原样返回，但实际使用过程中肯定不会这样使用。我在这里是希望通过在交互过程中集成GPT4o来让其更加智能化。
具体可以看下面的代码：
""" 大模型相关的封装和调用, 文件名是llms.py """ import os import requests import json import base64 from logger import logger # 读取图片并编码为 Base64 字符串 def encode_image_to_base64(image_path=None, image_bytes=None): if image_path and image_bytes: raise ValueError("image_path and image_bytes cannot be both provided.") if image_path: with open(image_path, "rb") as image_file: encoded_string = base64.b64encode(image_file.read()).decode("utf-8") if image_bytes: encoded_string = base64.b64encode(image_bytes).decode("utf-8") return encoded_string def azure_gpt4o(message): api_key = os.getenv("API_KEY") headers = {"Content-Type": "application/json", "api-key": api_key} data = { "messages": message, "max_tokens": 4096, "temperature": 0.8, "frequency_penalty": 0, "presence_penalty": 0, "top_p": 0.95, "stop": None, } url = os.environ.get("GPT4O_ENDPOINT") try: response = requests.post(url, headers=headers, data=json.dumps(data)) if response.status_code == 200: return response.json()["choices"][0]["message"]["content"] else: logger.info( f"LLM 调用失败，状态码：{response.status_code}，错误信息：{response.text}" ) except requests.RequestException as e: logger.error(f"请求发生错误：{e}") except Exception as e: logger.error(f"发生未知错误：{e}") 