AI大模型落地应用之电机驱动
一，关于将将AI大模型落地的一点思考。
2023年，随着ChatGPT的问世，大模型进入了告诉发展的时代。而这一年也被称为“人工智能元年”。
到了2024，大模型技术已经逐渐开始成熟，并试图进入到下一个阶段“大模型落地阶段”。
但是，到底该如何落地，怎么落地，这就成了一个问题。
毕竟，AI虽然可以赋能多个行业，但是它在某种意义上来说，应该算是虚拟产品。
因此，要实现他的落地，那我们就应该将他与实体产业结合起来。
而在中众多的实体产业中，机器人，无疑是最为契合的一个。
二，如何使用大模型技术来结合机器人技术？
相比于其他的硬件部分，属于机器人控制部分的软件部分应该是最为契合AI大模型使用的。
因此，在这里我主要以简单的AI驱动电机作为例子来进行一个说明。
1，准备工作
先去获取需要使用的API，OpenAI API官网链接：https://platform.openai.com/docs/api-reference
安装包含python环境的anaconda，它和直接安装python的好处是可以隔离多个环境，我们可以给每个项目创建一个纯净的环境，避免包之间的冲突。
Python
环境安装
安装anaconda
https://www.anaconda.com/products/individual
打开命令行，创建虚拟环境
conda create -n openai_control python=3.11
下载需要的包
pip install openai |
我这里使用的是pycharm，所以有些终端操作可以直接略过。
2，代码编写：
关于驱动电机控制的操作，由于我只是介绍一种AI落地应用的方向，所以我并没有搞太复杂。
基本上其核心内容就是：
大模型+Tool+串口通信+STM32控制。
1首先，我们先构造出ChatGPT对话的基本框架：
Python
import openai
api_key = open('api_key.txt', 'r').read().strip()
# 创建OpenAI的client，需要设置api_key
client = openai.OpenAI(api_key=api_key)
# 调用chat接口
chat_completion = client.chat.completions.create(
# 发送的消息内容，类似在ChatGPT中输入"hi，你好，请介绍下自己"
messages=[{
"role": "user",
"content": "hi，你好，请介绍下自己.",
}],
# gpt模型选择
model='gpt-3.5-turbo-1106',
# 非流式输出
stream=False
)
# GPT返回的答复
print(chat_completion.choices[0].message.content) |
3，构造电机驱动的Tools
Tools的本质就是插件。插件功能相当于给OpenAI增加了一个额外的扩展，开发者可以通过这个扩展提升它的能力。
调用流程如下：
1定义函数
定义本地函数Motor_control实现从API拉取，这里直接写一个简单对参数输出进行模拟。
1第一次调用接口
返回大模型分析出的函数名称和参数。
1调用本地函数
获取返回值，进行本地python方法调用
1第二次调用接口
将第一次接口的返回值message与本地函数调用的接口拼装起来，然后再次调用接口。
具体代码：
Python
import openai
# 自定义一个插件方法，用于进行电机的启动控制
def Motor_control():
return "启动"
# 定义messages
messages = [{
"role": "user",
"content": "hi，你好，你能启动一下电机吗？",
}]
api_key = open('api_key.txt', 'r').read().strip()
# 创建OpenAI的client，需要设置api_key
client = openai.OpenAI(api_key=api_key)
# 调用chat接口
first_chat_completion = client.chat.completions.create(
messages=messages,
model='gpt-3.5-turbo-1106',
stream=False,
# 开启插件
tool_choice="auto",
tools=[{
"type": "function",
"function": {
# tool的名字
"name": "Motor_control",
# tool的描述
"description": "控制电机的启动和停止",
}
}]
)
response_message = first_chat_completion.choices[0].message
print(response_message)
# 将GPT的答复记录下
messages.append(response_message)
# GPT返回的答复命中了插件
if response_message.tool_calls is not None:
print("命中插件，进行调用")
# 将结果放入到messages中
messages.append({
"role": "tool",
# 调用Func进行电机控制
"content": Motor_control(),
# 赋值GPT的返回中的tool call id，用于唯一标识一次tool call
# "tool_call_id": response_message.tool_calls[0].id
"tool_call_id" : response_message.tool_calls[0].id
})
# 发起二次请求
second_chat_completion = client.chat.completions.create(
messages=messages,
model='gpt-3.5-turbo-1106',
stream=False,
# 开启插件
tool_choice="auto",
tools=[{
"type": "function",
"function": {
# tool的名字
"name": "Motor_control",
# tool的描述
"description": "控制电机的启动和停止",
}
}]
)
print(second_chat_completion.choices[0].message.content) |
逻辑上看起来没有问题，并且对插件进行了命中，但是这里却报错了：
主要报错如下：
Python
raise self._make_status_error_from_response(err.response) from None
openai.BadRequestError: Error code: 400 - {'error': {'message': "An assistant message with 'tool_calls' must be followed by tool messages responding to each 'tool_call_id'. The following tool_call_ids did not have response messages: call_PEhO1CpC6DcW3QKQFacJqO2b", 'type': 'invalid_request_error', 'param': 'messages', 'code': None}}
|
根据后面查询方法，我发现这是因为tool_call_id的问题，
如果确定只有单个tool_call_id需要响应，那么代码应该只关注处理这一个响应。确保在向messages列表追加响应时，正确地引用了tool_call_id，并且提供了实际的响应内容。
但是，我的代码似乎中不止一次应用。
于是乎，据此，我进行了修正，下面是修正后的代码段：
Python
import openai
def Motor_control():
return "电机已启动"
# 定义messages
messages = [{
"role": "user",
"content": "hi，你好，你能启动一下电机吗？",
}]
api_key = open('api_key.txt', 'r').read().strip()
# 创建OpenAI的client，需要设置api_key
client = openai.OpenAI(api_key=api_key)
# 调用chat接口
first_chat_completion = client.chat.completions.create(
messages=messages,
model='gpt-3.5-turbo-1106',
stream=False,
# 开启插件
tool_choice="auto",
tools=[{
"type": "function",
"function": {
# tool的名字
"name": "Motor_control",
# tool的描述
"description": "控制电机的启动和停止",
}
}]
)
response_message = first_chat_completion.choices[0].message
print(response_message)
messages.append(response_message)
if response_message.tool_calls is not None:
print("命中插件，进行调用")
# 确保对每个tool_call_id都有响应
for tool_call in response_message.tool_calls:
tool_response_content = Motor_control() # 假设Motor_control是工具响应逻辑
messages.append({
"role": "tool",
"content": tool_response_content,
"tool_call_id": tool_call.id
})
# 发起二次请求，此时messages中已包含对tool_call的响应
second_chat_completion = client.chat.completions.create(
messages=messages,
model='gpt-3.5-turbo-1106',
stream=False,
tool_choice="auto",
tools=[{ # 这部分可以简化，因为你已经在messages中指定了tool_call的响应
"type": "function",
"function": {
"name": "Motor_control",
"description": "控制电机的启动和停止",
}
}]
)
print(second_chat_completion.choices[0].message.content) |
而，这次的结果则是成功调用：
4，在进行了基础的插件命中之后，就可以进行串口通信+STM32的控制操作了。
由于我本次只是为了进行电机的启动好停止，因此我们只需要对原本的Motor_control也就是电机控制函数进行修改即可。
当然，需要注意的是，由于这里还多了个串口通信的功能，因此我们还需要下载pyserial的包。
修改后的Motor_control函数：
Python
import serial
# 初始化串口，这里以COM3端口为例，波特率根据实际情况设定
ser = serial.Serial('COM3', 9600)
def Motor_control(user_input):
"""
根据用户输入决定发送启动(1)或停止(0)命令到串口。
:param user_input: 用户的原始输入文本
:return: 成功执行的响应信息
"""
if "启动" in user_input or "开" in user_input:
command = "1"
ser.write(command.encode()) # 发送启动命令
return "电机已启动"
elif "停止" in user_input or "关" in user_input:
command = "0"
ser.write(command.encode()) # 发送停止命令
return "电机已停止" |
此外，由于设计到了参数的加入，我们还需要对GPT的回复部分进行插件的参数重新注册：
Python
def interact_with_gpt():
completion = client.chat.completions.create(
messages=messages,
model='gpt-3.5-turbo-1106',
stream=False,
tool_choice="auto",
tools=[
{
"type": "function",
"function": {
"name": "Motor_control",
"description": "控制电机的启动和停止",
"parameters": {
"type": "object",
"properties": {
"command": {
"type": "string",
"description": "启动或停止电机的命令",
}
},
"required": ["command"]
},
}
},
]
)
return completion |
报错：
原因是因为：
tool_call.function.arguments 是一个字符串，而不是字典，所以你需要先将其转换为字典才能使用 get() 方法。 |
以下是修改后的代码：
Python
if response.tool_calls is not None:
print("命中插件，进行调用")
if response.tool_calls:
for tool_call in response.tool_calls:
arguments = json.loads(tool_call.function.arguments) # 将字符串转换为字典
if tool_call.function.name == "Motor_control":
# 直接处理电机控制逻辑，假设GPT会给出明确的命令
tool_response = Motor_control(arguments.get("command", ""))
else:
tool_response = {"content": "未知的工具调用。"}
messages.append({
"role": "tool",
"content": json.dumps(tool_response),
"tool_call_id": tool_call.id
}) |
至此，电机启动！
但是，有个问题是，我们的目的是通过串口来进行控制电机，也就是说我们还得要知道到底发生的串口数据是什么？
而GPT这里只告诉我们电机启动成功，但是没有告诉发送了什么数据。
于是我们再对电机控制函数进行修改，明确告诉GPT发送了什么数据：
Python
def Motor_control(user_input):
"""
根据用户输入决定发送启动(1)或停止(0)命令到串口。
:param user_input: 用户的原始输入文本
:return: 成功执行的响应信息
"""
if "启动" in user_input or "start" in user_input:
command = "1"
ser.write(command.encode()) # 发送启动命令
return "已发送数据1，电机已启动"
elif "停止" in user_input or "stop" in user_input:
command = "0"
ser.write(command.encode()) # 发送停止命令
return "已发送数据0，电机已停止" |
但是，效果却还是不行：
究其原因，是因为我们在询问的时候没有给出具体要求，我们的问题是：
因此，我们再对其进行修改：
而现在得到的结果就是我们想要的了：
5，编写STM32的控制代码
因为这里涉及的功能主要就是一个串口通信和PWM输出的功能。
所以我就主要展示这两部分：
C
#include "stm32f10x.h"
// 初始化USART1
void USART1_Init(void);
// 初始化TIM1 PWM
void TIM1_PWM_Init(void);
// 开始电机
void StartMotor(void);
// USART1中断服务例程
void USART1_IRQHandler(void);
int main(void)
{
NVIC_PriorityGroupConfig(NVIC_PriorityGroup_2);
USART1_Init();
TIM1_PWM_Init();
// 启动串口接收中断
NVIC_EnableIRQ(USART1_IRQn);
while (1)
{
// 主循环可以保持空闲，等待中断处理
}
}
// 初始化USART1
void USART1_Init(void)
{
GPIO_InitTypeDef GPIO_InitStructure;
USART_InitTypeDef USART_InitStructure;
NVIC_InitTypeDef NVIC_InitStructure;
RCC_APB2PeriphClockCmd(RCC_APB2Periph_USART1 | RCC_APB2Periph_GPIOA, ENABLE);
// 配置PA9（USART1_TX）和PA10（USART1_RX）
GPIO_InitStructure.GPIO_Pin = GPIO_Pin_9 | GPIO_Pin_10;
GPIO_InitStructure.GPIO_Mode = GPIO_Mode_AF_PP;
GPIO_InitStructure.GPIO_Speed = GPIO_Speed_50MHz;
GPIO_Init(GPIOA, &GPIO_InitStructure);
USART_InitStructure.USART_BaudRate = 9600;
USART_InitStructure.USART_WordLength = USART_WordLength_8b;
USART_InitStructure.USART_StopBits = USART_StopBits_1;
USART_InitStructure.USART_Parity = USART_Parity_No;
USART_InitStructure.USART_HardwareFlowControl = USART_HardwareFlowControl_None;
USART_InitStructure.USART_Mode = USART_Mode_Rx | USART_Mode_Tx;
USART_Init(USART1, &USART_InitStructure);
USART_Cmd(USART1, ENABLE);
NVIC_InitStructure.NVIC_IRQChannel = USART1_IRQn;
NVIC_InitStructure.NVIC_IRQChannelPreemptionPriority = 0;
NVIC_InitStructure.NVIC_IRQChannelSubPriority = 1;
NVIC_InitStructure.NVIC_IRQChannelCmd = ENABLE;
NVIC_Init(&NVIC_InitStructure);
}
// 初始化TIM1 PWM
void TIM1_PWM_Init(void)
{
TIM_TimeBaseInitTypeDef TIM_TimeBaseStructure;
TIM_OCInitTypeDef TIM_OCInitStructure;
RCC_APB2PeriphClockCmd(RCC_APB2Periph_TIM1, ENABLE);
TIM_TimeBaseStructure.TIM_Period = 1000; // 周期，根据系统时钟调整
TIM_TimeBaseStructure.TIM_Prescaler = 84 - 1; // 预分频，设置合适的PWM频率
TIM_TimeBaseStructure.TIM_ClockDivision = 0;
TIM_TimeBaseStructure.TIM_CounterMode = TIM_CounterMode_Up;
TIM_TimeBaseInit(TIM1, &TIM_TimeBaseStructure);
TIM_OCInitStructure.TIM_OCMode = TIM_OCMode_PWM1;
TIM_OCInitStructure.TIM_OutputState = TIM_OutputState_Enable;
TIM_OCInitStructure.TIM_Pulse = 0; // 初始占空比为0
TIM_OCInitStructure.TIM_OCPolarity = TIM_OCPolarity_High;
TIM_OC1Init(TIM1, &TIM_OCInitStructure);
TIM_OC1PreloadConfig(TIM1, TIM_OCPreload_Enable);
TIM_ARRPreloadConfig(TIM1, ENABLE);
TIM_Cmd(TIM1, ENABLE);
}
// 开始电机
void StartMotor(void)
{
// 设置PWM占空比以启动电机，例如设置为50%占空比
TIM1->CCR1 = TIM1->ARR / 2; // 假设ARR已设置，调整CCR1达到期望的占空比
}
// USART1中断服务例程
void USART1_IRQHandler(void)
{
if (USART_GetITStatus(USART1, USART_IT_RXNE) != RESET)
{
char receivedData = USART_ReceiveData(USART1);
if (receivedData == '1')
{
StartMotor();
}
}
} |
当然，这里是个伪代码，具体的配置参数还需要一些调整，不过大致流程基本就是如此。
粤ICP备17114055号
支持私有云部署
