我要投稿

AI大模型落地应用之电机驱动

发布日期：2024-06-18 22:39:39 浏览次数： 2917

作者：机器人与AI

微信搜一搜，关注“机器人与AI”

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();
        }
    }
}

当然，这里是个伪代码，具体的配置参数还需要一些调整，不过大致流程基本就是如此。

53AI，企业落地大模型首选服务商

产品：场景落地咨询+大模型应用平台+行业解决方案

承诺：免费POC验证，效果达标后再合作。零风险落地应用大模型，已交付160+中大型企业