ESP32-S3接入大模型API，对话AI
1、先使用python验证可行性
import requests url = "https://api.siliconflow.cn/v1/chat/completions" payload = { "model": "deepseek-ai/DeepSeek-R1", "messages": [ # 必须包含消息内容 {"role": "user", "content": "请解释量子计算的基本原理"} ], "stream": False, "max_tokens": 512, "temperature": 0.7, "top_p": 0.7, "top_k": 50, "frequency_penalty": 0.5, "n": 1 } headers = { "Authorization": "Bearer "your api_key"", "Content-Type": "application/json" } response = requests.request("POST", url, json=payload, headers=headers) print(response.text)  代码 |
|
|---|---|
|
秘钥获取 |
硅基流动注册送14元免费额度，https://cloud.siliconflow.cn/i/9MqV8tO4
邀请码：9MqV8tO4
2、Postman验证POST请求结构，深化熟悉对接流程
| POST请求头（Headers）设置 |
POST请求主体（Body）设置 |
|---|---|
| Authorization: your api key Content-Type：application/json |
Body:
{ "model": "deepseek-ai/DeepSeek-V3", "messages": [ {"role": "user", "content": "请解释量子计算的基本原理"}//问题 ], // "stream": False, "max_tokens": 512, "temperature": 0.7, "top_p": 0.7, "top_k": 50, "frequency_penalty": 0.5, "n": 1 } 实战！在esp32s3上实现ai对话：
1、使用micro python实现
因为之前验证过python的可行性，而esp32系列可以使用micro python编程，其特点是便捷高效，所以先使用mico python实现。
开发环境
编程调试下载环境：Thonny
编程语言：micro python
前置条件：烧录micro python固件（乐鑫官网中下载）
import network import time import urequests import ujson from machine import reset # ====== 配置部分 ====== SSID = 'jianzhiji' PASSWORD = '8765432111' API_KEY = "你的api密钥" API_URL = "https://api.siliconflow.cn/v1/chat/completions" def connect_wifi(): wlan = network.WLAN(network.STA_IF) wlan.active(True) if not wlan.isconnected(): print(f"Connecting to {SSID}...") wlan.connect(SSID, PASSWORD) timeout = 20 while not wlan.isconnected() and timeout > 0: print(".", end="") time.sleep(1) timeout -= 1 if not wlan.isconnected(): raise RuntimeError("WiFi连接超时") print("nIP Address:", wlan.ifconfig()[0]) return wlan def api_request(): headers = { "Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json" } # 最小化请求参数 payload = { "model": "deepseek-ai/DeepSeek-V3", "messages": [{"role": "user", "content": "hello"}], "max_tokens": 512, "temperature": 0.7, "top_p": 0.7 } try: print("n[Request]") print("Payload:", payload) response = urequests.post( API_URL, headers=headers, data=ujson.dumps(payload), timeout=20 ) print(f"n[Response] Status: {response.status_code}") if response.status_code == 200: json_resp = response.json() print("AI回复:", json_resp['choices'][0]['message']['content']) else: print("Error Response:", response.text) response.close() except Exception as e: print("[Error]", str(e)) try: print("=== 启动系统 ===") connect_wifi() api_request() except Exception as e: print("n!!! 错误:", str(e)) print("10秒后重启...") time.sleep(10) reset()
2、使用C语言实现
开发环境
编程调试下载环境：VSCOde+espidf插件
编程语言：C
#include <string.h> #include "esp_log.h" #include "nvs_flash.h" #include "esp_event.h" #include "esp_netif.h" #include "protocol_examples_common.h" #include "esp_http_client.h" #include "esp_sntp.h" // 在文件开头添加 #include "lwip/apps/sntp.h" #define MAX_HTTP_OUTPUT_BUFFER 2048 static const char *TAG = "HTTP_AI_CLIENT"; #define API_ENDPOINT "http://api.siliconflow.cn/v1/chat/completions" // 添加事件处理函数 static esp_err_t http_event_handler(esp_http_client_event_t *evt) { switch (evt->event_id) { case HTTP_EVENT_ERROR: ESP_LOGI(TAG, "HTTP_EVENT_ERROR"); break; case HTTP_EVENT_ON_CONNECTED: ESP_LOGI(TAG, "HTTP_EVENT_ON_CONNECTED"); break; case HTTP_EVENT_HEADERS_SENT: ESP_LOGI(TAG, "HTTP_EVENT_HEADERS_SENT"); break; case HTTP_EVENT_ON_HEADER: ESP_LOGI(TAG, "HTTP_EVENT_ON_HEADER, key=%s, value=%s", evt->header_key, evt->header_value); break; case HTTP_EVENT_ON_DATA: ESP_LOGI(TAG, "HTTP_EVENT_ON_DATA, len=%d", evt->data_len); break; case HTTP_EVENT_ON_FINISH: ESP_LOGI(TAG, "HTTP_EVENT_ON_FINISH"); break; case HTTP_EVENT_DISCONNECTED: ESP_LOGI(TAG, "HTTP_EVENT_DISCONNECTED"); break; default: break; } return ESP_OK; } static void http_post_request(void) { char output_buffer[MAX_HTTP_OUTPUT_BUFFER] = {0}; esp_http_client_config_t config = { .url = API_ENDPOINT, .event_handler = http_event_handler, // 添加事件处理 .timeout_ms = 30000, // 增加超时到30秒 .skip_cert_common_name_check = true, // 跳过证书CN检查 .cert_pem = NULL, // 不指定证书 }; esp_http_client_handle_t client = esp_http_client_init(&config); const char *json_data = "{" ""model": "deepseek-ai/DeepSeek-V3","//选择模型 ""messages": [" "{" ""role": "user"," ""content": "你好""//填入问题内容 "}" "]," ""max_tokens": 512," ""temperature": 0.7," ""top_p": 0.7," ""top_k": 50," ""frequency_penalty": 0.5," ""n": 1" "}"; esp_http_client_set_method(client, HTTP_METHOD_POST); esp_http_client_set_header(client, "Content-Type", "application/json"); esp_http_client_set_header(client, "Authorization", "your api key");//key_API esp_err_t err = esp_http_client_open(client, strlen(json_data)); if (err != ESP_OK) { ESP_LOGE(TAG, "连接失败: %s", esp_err_to_name(err)); goto cleanup; } int wlen = esp_http_client_write(client, json_data, strlen(json_data)); if (wlen < 0) { ESP_LOGE(TAG, "数据写入失败"); goto cleanup; } int content_length = esp_http_client_fetch_headers(client); if (content_length < 0) { ESP_LOGE(TAG, "获取头部信息失败"); goto cleanup; } int data_read = esp_http_client_read_response(client, output_buffer, MAX_HTTP_OUTPUT_BUFFER); if (data_read >= 0) { ESP_LOGI(TAG, "HTTP状态码 = %d, 内容长度 = %"PRIu64, esp_http_client_get_status_code(client), esp_http_client_get_content_length(client)); ESP_LOGI(TAG, "响应内容: %s", output_buffer); } else { ESP_LOGE(TAG, "读取响应失败"); } cleanup: esp_http_client_cleanup(client); } static void http_task(void *pvParameters) { http_post_request(); ESP_LOGI(TAG, "HTTP示例完成"); vTaskDelete(NULL); } void app_main(void) { ESP_ERROR_CHECK(nvs_flash_init()); ESP_ERROR_CHECK(esp_netif_init()); ESP_ERROR_CHECK(esp_event_loop_create_default()); ESP_ERROR_CHECK(example_connect()); sntp_setoperatingmode(SNTP_OPMODE_POLL); sntp_setservername(0, "pool.ntp.org"); // 第1个NTP服务器 sntp_setservername(1, "time.nist.gov"); // 第2个NTP服务器（可选） sntp_init(); // 在 sntp_init() 后添加时间同步等待 int retry = 0; const int retry_count = 10; while (sntp_get_sync_status() == SNTP_SYNC_STATUS_RESET && ++retry < retry_count) { printf("Waiting for system time to sync... (%d/%d)n", retry, retry_count); vTaskDelay(2000 / portTICK_PERIOD_MS); } xTaskCreate(&http_task, "http_task", 8192, NULL, 5, NULL); } 终端输出结果
由于该版本等待时间过短，提问复杂问题时常常等不到回答就提示连接失败。
以下优化版本增加
1、超时策略
2、动态缓冲区扩容
3、分段传输机制
#include <string.h> #include "esp_log.h" #include "nvs_flash.h" #include "esp_event.h" #include "esp_netif.h" #include "protocol_examples_common.h" #include "esp_http_client.h" #include "esp_sntp.h" // 在文件开头添加 #include "lwip/apps/sntp.h" #define MAX_HTTP_OUTPUT_BUFFER 2048 static const char *TAG = "HTTP_AI_CLIENT"; #define API_ENDPOINT "http://api.siliconflow.cn/v1/chat/completions" // 添加事件处理函数 static esp_err_t http_event_handler(esp_http_client_event_t *evt) { switch (evt->event_id) { case HTTP_EVENT_ERROR: ESP_LOGI(TAG, "HTTP_EVENT_ERROR"); break; case HTTP_EVENT_ON_CONNECTED: ESP_LOGI(TAG, "HTTP_EVENT_ON_CONNECTED"); break; case HTTP_EVENT_HEADERS_SENT: ESP_LOGI(TAG, "HTTP_EVENT_HEADERS_SENT"); break; case HTTP_EVENT_ON_HEADER: ESP_LOGI(TAG, "HTTP_EVENT_ON_HEADER, key=%s, value=%s", evt->header_key, evt->header_value); break; case HTTP_EVENT_ON_DATA: ESP_LOGI(TAG, "HTTP_EVENT_ON_DATA, len=%d", evt->data_len); break; case HTTP_EVENT_ON_FINISH: ESP_LOGI(TAG, "HTTP_EVENT_ON_FINISH"); break; case HTTP_EVENT_DISCONNECTED: ESP_LOGI(TAG, "HTTP_EVENT_DISCONNECTED"); break; default: break; } return ESP_OK; } static void http_post_request(void) { // char output_buffer[MAX_HTTP_OUTPUT_BUFFER] = {0}; // 增大缓冲区防止数据截断 char *output_buffer = calloc(1, MAX_HTTP_OUTPUT_BUFFER * 2); // 扩容至4096 if (!output_buffer) { ESP_LOGE(TAG, "内存分配失败"); return; } esp_http_client_config_t config = { .url = API_ENDPOINT, .event_handler = http_event_handler, .timeout_ms = 120000, // 超时延长至120秒 .disable_auto_redirect = false, // 允许重定向 .keep_alive_enable = true, // 启用长连接 .skip_cert_common_name_check = true, .cert_pem = NULL }; esp_http_client_handle_t client = esp_http_client_init(&config); const char *json_data = "{" ""model": "deepseek-ai/DeepSeek-R1","//选择模型 ""messages": [" "{" ""role": "user"," ""content": "写一篇散文""//填入问题内容 "}" "]," ""max_tokens": 512," ""temperature": 0.7," ""top_p": 0.7," ""top_k": 50," ""frequency_penalty": 0.5," ""n": 1" "}"; esp_http_client_set_method(client, HTTP_METHOD_POST); esp_http_client_set_header(client, "Content-Type", "application/json"); esp_http_client_set_header(client, "Authorization", " ");//key_API // 新增分段接收逻辑 int total_read = 0; esp_err_t err = esp_http_client_open(client, strlen(json_data)); if (err == ESP_OK) { // 分段写入数据（应对大请求体） const char *ptr = json_data; int remaining = strlen(json_data); while (remaining > 0) { int written = esp_http_client_write(client, ptr, remaining); if (written <= 0) break; ptr += written; remaining -= written; } } int wlen = esp_http_client_write(client, json_data, strlen(json_data)); if (wlen < 0) { ESP_LOGE(TAG, "数据写入失败"); goto cleanup; } // 新增分段读取逻辑 if (esp_http_client_fetch_headers(client) >= 0) { int read_len; do { read_len = esp_http_client_read(client, output_buffer + total_read, MAX_HTTP_OUTPUT_BUFFER * 2 - total_read - 1 ); if (read_len > 0) { total_read += read_len; } } while (read_len > 0); output_buffer[total_read] = ' '; // 确保字符串终止 } int data_read = esp_http_client_read_response(client, output_buffer, MAX_HTTP_OUTPUT_BUFFER); if (data_read >= 0) { ESP_LOGI(TAG, "HTTP状态码 = %d, 内容长度 = %"PRIu64, esp_http_client_get_status_code(client), esp_http_client_get_content_length(client)); ESP_LOGI(TAG, "响应内容: %s", output_buffer); } else { ESP_LOGE(TAG, "读取响应失败"); } cleanup: esp_http_client_cleanup(client); } static void http_task(void *pvParameters) { http_post_request(); ESP_LOGI(TAG, "HTTP示例完成"); vTaskDelete(NULL); } void app_main(void) { ESP_ERROR_CHECK(nvs_flash_init()); ESP_ERROR_CHECK(esp_netif_init()); ESP_ERROR_CHECK(esp_event_loop_create_default()); ESP_ERROR_CHECK(example_connect()); sntp_setoperatingmode(SNTP_OPMODE_POLL); sntp_setservername(0, "pool.ntp.org"); // 第1个NTP服务器 sntp_setservername(1, "time.nist.gov"); // 第2个NTP服务器（可选） sntp_init(); // 在 sntp_init() 后添加时间同步等待 int retry = 0; const int retry_count = 10; while (sntp_get_sync_status() == SNTP_SYNC_STATUS_RESET && ++retry < retry_count) { printf("Waiting for system time to sync... (%d/%d)n", retry, retry_count); vTaskDelay(2000 / portTICK_PERIOD_MS); } xTaskCreate(&http_task, "http_task", 18192, NULL, 5, NULL); } 终端打印接收结果，可以成功等待接受长文本
NTP服务器
NTP服务器（Network Time Protocol Server）的作用是提供精确、统一的时间同步服务，它是互联网中时间同步的核心基础设施。以下是其核心作用：
1. 时间同步
- 核心功能：NTP服务器通过协议将设备（如计算机、手机、物联网设备）的本地时间与全球标准时间（如UTC）同步。
- 解决设备时钟漂移：硬件时钟可能因温度、电池等原因产生误差，NTP定期校准时间，确保设备间时间一致。
2. 安全协议的基础
- SSL/TLS证书验证：HTTPS通信依赖证书的有效期验证。若设备时间错误：
- 证书可能被误判为“未生效”或“已过期”，导致连接失败。
- 示例场景：你之前的ESP32代码中，若系统时间未同步，访问HTTPS API时可能因证书时间校验失败而超时。
3. 日志与调试
- 统一时间戳：设备生成的日志若时间不一致，将难以追踪问题（如分布式系统故障）。
- 示例：ESP32的日志若时间错误，无法与服务器日志对照，增加调试难度。
4. 协调分布式系统
- 跨设备协同：物联网、金融交易等场景依赖毫秒级时间同步，避免数据冲突。
- 示例：多个ESP32设备协同工作时，需严格按时间顺序执行操作。
5. NTP与SNTP的区别
- NTP：高精度（毫秒/微秒级），适用于服务器、基站等。
- SNTP（简单NTP）：精简版，精度稍低（秒级），适合资源受限设备（如ESP32）。
- 你的代码：使用
esp_sntp_init()即是为ESP32提供轻量级时间同步。
为什么你的ESP32项目需要NTP服务器？
在之前的代码中，你遇到HTTP连接超时问题，可能原因之一是：
- ESP32系统时间未同步 → SSL证书验证失败 → 服务器拒绝连接。
- 初始化SNTP后，设备获取正确时间 → 证书验证通过 → 正常通信。
总结
NTP服务器像互联网的“原子钟”，确保设备时间精准统一。对物联网设备而言，它是安全通信、协同工作的基石。
