Qwen2.5-VL-Chord视觉定位模型API调用详解requests POST方式远程调用示例1. 项目简介什么是视觉定位想象一下你给朋友发了一张照片然后问“照片里那个红色的杯子在哪”朋友会告诉你“在桌子左边靠近窗户的位置。”这个过程就是视觉定位——用语言描述在图像中找到具体的东西。Qwen2.5-VL-Chord就是这样一个智能的“视觉定位专家”。它基于Qwen2.5-VL多模态大模型能够理解你的文字描述然后在图片里精准找到对应的物体并用一个方框bounding box标出来。1.1 核心能力它能做什么这个模型最厉害的地方在于听懂人话找东西你说“找到图里的白色花瓶”它就能在图片里找到白色花瓶的位置支持多种目标日常物品、人像、场景元素都能找不需要提前标注数据返回精确坐标不只是告诉你“有”而是给出具体的像素坐标位置远程调用方便通过简单的API接口任何程序都能调用这个能力1.2 实际应用场景这个技术在实际工作中特别有用电商平台用户上传商品图片自动识别和定位商品主体内容审核在图片中定位敏感内容或违规物品智能相册根据描述快速找到特定照片中的某个元素工业质检定位产品缺陷或特定部件的位置机器人视觉让机器人“看懂”指令找到目标物体2. 为什么需要远程API调用你可能会有疑问既然有Web界面为什么还要学API调用2.1 Web界面的局限性Web界面确实方便点点鼠标就能用。但在实际项目中你可能会遇到这些问题批量处理困难一次只能处理一张图片几百张图片要手动操作集成到系统难无法直接嵌入到你的应用程序中自动化流程中断需要人工介入无法实现全自动处理性能监控不便难以统计调用次数、成功率等指标2.2 API调用的优势相比之下API调用就像给你的程序装上了“眼睛”批量自动化一次调用可以处理多张图片完全自动化系统集成轻松嵌入到现有业务流程中灵活定制可以根据需求调整参数优化效果性能可控可以监控响应时间、成功率等指标3. 环境准备调用前需要什么在开始调用API之前我们先确保环境都准备好了。3.1 确认服务正常运行首先确保Chord服务已经在服务器上正常运行# 登录到部署Chord服务的服务器 ssh useryour-server-ip # 检查服务状态 supervisorctl status chord如果看到类似下面的输出说明服务正常chord RUNNING pid 135976, uptime 0:01:343.2 获取服务地址和端口你需要知道服务的访问地址服务器IP部署Chord服务的服务器地址端口号默认是7860可以在配置中修改API地址通常是http://服务器IP:7860/api/predict3.3 准备测试图片准备一些测试图片建议从简单到复杂简单场景单个人物或物体的清晰图片中等场景多个物体背景不太复杂复杂场景多个物体背景复杂有遮挡4. 基础调用最简单的POST请求让我们从一个最简单的例子开始看看如何用Python的requests库调用这个API。4.1 完整代码示例import requests import base64 from PIL import Image import io def simple_chord_api_call(image_path, prompt, server_urlhttp://localhost:7860): 最简单的Chord API调用函数 参数 image_path: 图片文件路径 prompt: 文本提示如找到图中的人 server_url: Chord服务地址默认本地7860端口 # 1. 读取图片并转换为base64 with open(image_path, rb) as image_file: image_bytes image_file.read() image_base64 base64.b64encode(image_bytes).decode(utf-8) # 2. 准备请求数据 payload { data: [ image_base64, # 图片的base64编码 prompt, # 文本提示 512, # max_new_tokens参数 0.6, # temperature参数 0.9, # top_p参数 1.0 # repetition_penalty参数 ] } # 3. 发送POST请求 try: response requests.post( f{server_url}/api/predict, jsonpayload, timeout30 # 设置30秒超时 ) # 4. 检查响应状态 if response.status_code 200: result response.json() return result else: print(f请求失败状态码{response.status_code}) print(f错误信息{response.text}) return None except requests.exceptions.RequestException as e: print(f请求异常{e}) return None # 使用示例 if __name__ __main__: # 调用API result simple_chord_api_call( image_pathtest.jpg, prompt找到图中的人, server_urlhttp://192.168.1.100:7860 # 替换为你的服务器地址 ) if result: print(API调用成功) print(f返回数据{result})4.2 代码逐行解释让我详细解释一下这个代码的每个部分第1步图片处理with open(image_path, rb) as image_file: image_bytes image_file.read() image_base64 base64.b64encode(image_bytes).decode(utf-8)这里做了三件事以二进制模式打开图片文件读取图片的字节数据将字节数据转换为base64字符串因为HTTP请求不能直接发送二进制文件第2步准备请求数据payload { data: [ image_base64, # 图片 prompt, # 文本提示 512, # 生成的最大token数 0.6, # 温度参数控制随机性 0.9, # top_p参数控制多样性 1.0 # 重复惩罚避免重复内容 ] }这个结构是Gradio API的标准格式data列表中的每个位置都有特定含义。第3步发送请求response requests.post( f{server_url}/api/predict, jsonpayload, timeout30 )使用POST方法发送请求数据以JSON格式发送设置30秒超时避免长时间等待第4步处理响应if response.status_code 200: result response.json() return result状态码200表示成功将响应的JSON数据解析为Python字典4.3 运行测试保存上面的代码为chord_api_simple.py然后运行python chord_api_simple.py如果一切正常你会看到类似这样的输出API调用成功 返回数据{data: [处理后的base64图片, 找到1个人位置在...], ...}5. 进阶功能处理复杂场景基础调用会了现在我们来看看更实用的进阶功能。5.1 批量处理多张图片在实际项目中我们经常需要处理大量图片。手动一张张调用太慢了看看如何批量处理import requests import base64 import os from concurrent.futures import ThreadPoolExecutor, as_completed import time class ChordBatchProcessor: 批量处理Chord API调用的类 def __init__(self, server_url, max_workers3): self.server_url server_url self.max_workers max_workers # 最大并发数 def process_single_image(self, image_path, prompt): 处理单张图片 try: # 读取图片 with open(image_path, rb) as f: image_base64 base64.b64encode(f.read()).decode(utf-8) # 准备请求 payload { data: [image_base64, prompt, 512, 0.6, 0.9, 1.0] } # 发送请求 response requests.post( f{self.server_url}/api/predict, jsonpayload, timeout60 ) if response.status_code 200: result response.json() return { status: success, image: os.path.basename(image_path), result: result } else: return { status: error, image: os.path.basename(image_path), error: fHTTP {response.status_code} } except Exception as e: return { status: error, image: os.path.basename(image_path), error: str(e) } def process_batch(self, image_dir, prompt, image_extensions[.jpg, .png, .jpeg]): 批量处理目录中的所有图片 # 收集所有图片文件 image_files [] for file in os.listdir(image_dir): if any(file.lower().endswith(ext) for ext in image_extensions): image_files.append(os.path.join(image_dir, file)) print(f找到 {len(image_files)} 张图片需要处理) results [] start_time time.time() # 使用线程池并发处理 with ThreadPoolExecutor(max_workersself.max_workers) as executor: # 提交所有任务 future_to_image { executor.submit(self.process_single_image, img_path, prompt): img_path for img_path in image_files } # 收集结果 for future in as_completed(future_to_image): result future.result() results.append(result) # 实时显示进度 success_count sum(1 for r in results if r[status] success) print(f进度: {len(results)}/{len(image_files)}成功: {success_count}) # 统计信息 end_time time.time() total_time end_time - start_time success_results [r for r in results if r[status] success] error_results [r for r in results if r[status] error] print(f\n批量处理完成) print(f总耗时: {total_time:.2f}秒) print(f平均每张: {total_time/len(image_files):.2f}秒) print(f成功: {len(success_results)}张失败: {len(error_results)}张) return results # 使用示例 if __name__ __main__: # 创建处理器 processor ChordBatchProcessor( server_urlhttp://192.168.1.100:7860, max_workers3 # 根据服务器性能调整 ) # 批量处理 results processor.process_batch( image_dir./images, # 图片目录 prompt找到图中的人 # 统一的提示词 ) # 保存结果 import json with open(batch_results.json, w, encodingutf-8) as f: json.dump(results, f, ensure_asciiFalse, indent2)5.2 解析返回的边界框坐标API返回的数据中包含边界框信息我们需要正确解析def parse_bounding_boxes(api_response): 解析API返回的边界框信息 参数 api_response: API返回的完整响应数据 返回 解析后的边界框列表和图片尺寸 try: # 从响应中提取文本结果 result_data api_response.get(data, []) if len(result_data) 2: return None text_result result_data[1] # 文本结果在第二个位置 # 解析边界框格式示例box(100,150,200,300)/box boxes [] # 方法1使用正则表达式提取box标签内容 import re box_pattern rbox\((.*?)\)/box matches re.findall(box_pattern, text_result) for match in matches: # 解析坐标字符串如 100,150,200,300 coords [int(coord.strip()) for coord in match.split(,)] if len(coords) 4: boxes.append({ x1: coords[0], # 左上角x y1: coords[1], # 左上角y x2: coords[2], # 右下角x y2: coords[3], # 右下角y width: coords[2] - coords[0], # 宽度 height: coords[3] - coords[1] # 高度 }) # 方法2如果API直接返回结构化数据 if boxes in api_response: boxes api_response[boxes] # 获取图片尺寸如果有 image_size None if image_size in api_response: image_size api_response[image_size] return { text: text_result, boxes: boxes, image_size: image_size, count: len(boxes) } except Exception as e: print(f解析边界框时出错{e}) return None # 使用示例 if __name__ __main__: # 假设这是API返回的数据 api_response { data: [ base64_encoded_image_data, 找到2个人位置分别是box(100,150,200,300)/box和box(250,180,350,400)/box ] } parsed_result parse_bounding_boxes(api_response) if parsed_result: print(f找到 {parsed_result[count]} 个目标) for i, box in enumerate(parsed_result[boxes]): print(f目标{i1}: 位置({box[x1]},{box[y1]})到({box[x2]},{box[y2]}) f大小{box[width]}x{box[height]})5.3 可视化标注结果解析出坐标后我们可以在本地图片上绘制边界框直观地查看结果from PIL import Image, ImageDraw, ImageFont import matplotlib.pyplot as plt def visualize_bounding_boxes(image_path, boxes, save_pathNone): 在图片上绘制边界框并显示 参数 image_path: 原始图片路径 boxes: 边界框列表每个元素是[x1, y1, x2, y2] save_path: 保存结果图片的路径可选 # 打开图片 image Image.open(image_path) draw ImageDraw.Draw(image) # 定义颜色可以循环使用不同颜色 colors [red, blue, green, orange, purple] # 绘制每个边界框 for i, box in enumerate(boxes): color colors[i % len(colors)] # 绘制矩形框 draw.rectangle( [box[x1], box[y1], box[x2], box[y2]], outlinecolor, width3 ) # 添加标签 label f目标{i1} try: # 尝试使用字体 font ImageFont.truetype(arial.ttf, 20) except: # 回退到默认字体 font ImageFont.load_default() # 计算文字位置在框的上方 text_bbox draw.textbbox((0, 0), label, fontfont) text_width text_bbox[2] - text_bbox[0] text_height text_bbox[3] - text_bbox[1] # 绘制文字背景 draw.rectangle( [box[x1], box[y1] - text_height - 5, box[x1] text_width 10, box[y1]], fillcolor ) # 绘制文字 draw.text( (box[x1] 5, box[y1] - text_height - 2), label, fillwhite, fontfont ) # 显示图片 plt.figure(figsize(12, 8)) plt.imshow(image) plt.axis(off) plt.title(f检测到 {len(boxes)} 个目标) plt.show() # 保存图片如果指定了保存路径 if save_path: image.save(save_path) print(f结果已保存到: {save_path}) return image # 使用示例 if __name__ __main__: # 假设我们已经从API获取了边界框 boxes [ {x1: 100, y1: 150, x2: 200, y2: 300}, {x1: 250, y1: 180, x2: 350, y2: 400} ] # 可视化结果 result_image visualize_bounding_boxes( image_pathtest.jpg, boxesboxes, save_pathtest_with_boxes.jpg )6. 实战案例完整的项目集成现在我们把所有功能整合起来创建一个完整的项目示例。6.1 项目结构设计vision_grounding_project/ ├── config/ │ └── settings.py # 配置文件 ├── src/ │ ├── chord_client.py # Chord API客户端 │ ├── image_processor.py # 图片处理工具 │ ├── result_parser.py # 结果解析器 │ └── visualizer.py # 可视化工具 ├── tests/ │ └── test_api.py # 测试脚本 ├── data/ │ ├── input/ # 输入图片 │ └── output/ # 输出结果 ├── logs/ # 日志文件 ├── requirements.txt # 依赖包 └── main.py # 主程序6.2 完整的API客户端类# src/chord_client.py import requests import base64 import time import json from typing import List, Dict, Optional, Tuple import logging class ChordAPIClient: Chord API客户端封装所有API调用功能 def __init__(self, server_url: str, timeout: int 60): 初始化客户端 参数 server_url: Chord服务地址如 http://192.168.1.100:7860 timeout: 请求超时时间秒 self.server_url server_url.rstrip(/) self.timeout timeout self.api_url f{self.server_url}/api/predict # 设置日志 logging.basicConfig( levellogging.INFO, format%(asctime)s - %(name)s - %(levelname)s - %(message)s ) self.logger logging.getLogger(__name__) # 测试连接 self._test_connection() def _test_connection(self): 测试与服务器的连接 try: response requests.get(self.server_url, timeout10) if response.status_code 200: self.logger.info(f成功连接到Chord服务: {self.server_url}) return True else: self.logger.warning(f连接测试返回状态码: {response.status_code}) return False except Exception as e: self.logger.error(f连接测试失败: {e}) return False def encode_image_to_base64(self, image_path: str) - Optional[str]: 将图片文件编码为base64字符串 try: with open(image_path, rb) as f: image_bytes f.read() return base64.b64encode(image_bytes).decode(utf-8) except Exception as e: self.logger.error(f图片编码失败 {image_path}: {e}) return None def call_api(self, image_base64: str, prompt: str, max_new_tokens: int 512, temperature: float 0.6, top_p: float 0.9, repetition_penalty: float 1.0) - Optional[Dict]: 调用Chord API 参数 image_base64: 图片的base64编码字符串 prompt: 文本提示 其他参数模型生成参数 返回 API响应数据字典失败返回None # 准备请求数据 payload { data: [ image_base64, prompt, max_new_tokens, temperature, top_p, repetition_penalty ] } # 记录请求开始时间 start_time time.time() try: self.logger.info(f发送API请求提示词: {prompt[:50]}...) response requests.post( self.api_url, jsonpayload, timeoutself.timeout, headers{Content-Type: application/json} ) # 计算耗时 elapsed_time time.time() - start_time if response.status_code 200: result response.json() self.logger.info(fAPI调用成功耗时: {elapsed_time:.2f}秒) return { success: True, data: result, elapsed_time: elapsed_time, status_code: response.status_code } else: self.logger.error( fAPI调用失败状态码: {response.status_code}, f耗时: {elapsed_time:.2f}秒 ) return { success: False, error: fHTTP {response.status_code}, response_text: response.text[:200], elapsed_time: elapsed_time, status_code: response.status_code } except requests.exceptions.Timeout: elapsed_time time.time() - start_time self.logger.error(fAPI请求超时耗时: {elapsed_time:.2f}秒) return { success: False, error: 请求超时, elapsed_time: elapsed_time } except Exception as e: elapsed_time time.time() - start_time self.logger.error(fAPI请求异常: {e}耗时: {elapsed_time:.2f}秒) return { success: False, error: str(e), elapsed_time: elapsed_time } def process_image(self, image_path: str, prompt: str, **kwargs) - Optional[Dict]: 处理单张图片的完整流程 参数 image_path: 图片文件路径 prompt: 文本提示 **kwargs: 其他API参数 返回 处理结果字典 # 1. 编码图片 self.logger.info(f开始处理图片: {image_path}) image_base64 self.encode_image_to_base64(image_path) if not image_base64: return { success: False, error: 图片编码失败, image_path: image_path } # 2. 调用API api_result self.call_api(image_base64, prompt, **kwargs) if not api_result or not api_result.get(success): return { success: False, error: api_result.get(error, API调用失败), image_path: image_path, api_result: api_result } # 3. 返回完整结果 return { success: True, image_path: image_path, prompt: prompt, api_response: api_result[data], elapsed_time: api_result[elapsed_time], timestamp: time.strftime(%Y-%m-%d %H:%M:%S) } def batch_process(self, image_paths: List[str], prompt: str, max_workers: int 3, **kwargs) - List[Dict]: 批量处理多张图片 参数 image_paths: 图片路径列表 prompt: 文本提示所有图片使用相同的提示 max_workers: 最大并发数 **kwargs: 其他API参数 返回 处理结果列表 from concurrent.futures import ThreadPoolExecutor, as_completed self.logger.info(f开始批量处理 {len(image_paths)} 张图片) results [] completed 0 start_time time.time() # 使用线程池并发处理 with ThreadPoolExecutor(max_workersmax_workers) as executor: # 提交所有任务 future_to_image { executor.submit(self.process_image, img_path, prompt, **kwargs): img_path for img_path in image_paths } # 收集结果 for future in as_completed(future_to_image): result future.result() results.append(result) completed 1 # 记录进度 if completed % 5 0 or completed len(image_paths): elapsed time.time() - start_time avg_time elapsed / completed if completed 0 else 0 self.logger.info( f进度: {completed}/{len(image_paths)} f平均耗时: {avg_time:.2f}秒/张 ) # 统计信息 total_time time.time() - start_time success_count sum(1 for r in results if r.get(success)) self.logger.info( f批量处理完成总共耗时: {total_time:.2f}秒 f成功: {success_count}/{len(image_paths)} ) return results def save_results(self, results: List[Dict], output_file: str): 保存处理结果到JSON文件 try: with open(output_file, w, encodingutf-8) as f: json.dump(results, f, ensure_asciiFalse, indent2) self.logger.info(f结果已保存到: {output_file}) return True except Exception as e: self.logger.error(f保存结果失败: {e}) return False # 使用示例 if __name__ __main__: # 创建客户端 client ChordAPIClient( server_urlhttp://192.168.1.100:7860, timeout60 ) # 处理单张图片 result client.process_image( image_pathtest.jpg, prompt找到图中的人, max_new_tokens512, temperature0.6 ) if result and result[success]: print(处理成功) print(f耗时: {result[elapsed_time]:.2f}秒) # 保存结果 client.save_results([result], single_result.json)6.3 主程序示例# main.py import os import sys from src.chord_client import ChordAPIClient from src.result_parser import parse_bounding_boxes from src.visualizer import visualize_results import argparse def main(): 主程序 # 解析命令行参数 parser argparse.ArgumentParser(descriptionChord视觉定位API客户端) parser.add_argument(--server, typestr, defaulthttp://localhost:7860, helpChord服务器地址) parser.add_argument(--image, typestr, help单张图片路径) parser.add_argument(--dir, typestr, help图片目录路径批量处理) parser.add_argument(--prompt, typestr, requiredTrue, help文本提示如找到图中的人) parser.add_argument(--output, typestr, defaultoutput, help输出目录) parser.add_argument(--workers, typeint, default3, help并发工作线程数) args parser.parse_args() # 创建输出目录 os.makedirs(args.output, exist_okTrue) # 创建客户端 print(f连接到Chord服务: {args.server}) client ChordAPIClient(server_urlargs.server) results [] if args.image: # 处理单张图片 print(f处理单张图片: {args.image}) result client.process_image(args.image, args.prompt) if result: results.append(result) # 解析和可视化 if result[success]: parsed parse_bounding_boxes(result[api_response]) if parsed: # 保存解析结果 import json with open(os.path.join(args.output, parsed_result.json), w) as f: json.dump(parsed, f, indent2) # 可视化 visualize_results( image_pathargs.image, boxesparsed[boxes], save_pathos.path.join(args.output, visualized.jpg) ) elif args.dir: # 批量处理 if not os.path.exists(args.dir): print(f目录不存在: {args.dir}) return # 收集图片文件 image_extensions [.jpg, .jpeg, .png, .bmp, .webp] image_files [] for root, dirs, files in os.walk(args.dir): for file in files: if any(file.lower().endswith(ext) for ext in image_extensions): image_files.append(os.path.join(root, file)) if not image_files: print(f在目录 {args.dir} 中未找到图片文件) return print(f找到 {len(image_files)} 张图片开始批量处理...) # 批量处理 results client.batch_process( image_pathsimage_files[:50], # 限制前50张避免太多 promptargs.prompt, max_workersargs.workers ) else: print(请指定 --image 或 --dir 参数) return # 保存所有结果 if results: output_file os.path.join(args.output, all_results.json) client.save_results(results, output_file) # 生成统计报告 success_count sum(1 for r in results if r.get(success)) total_time sum(r.get(elapsed_time, 0) for r in results if r.get(success)) print(\n *50) print(处理完成) print(f总共处理: {len(results)} 张图片) print(f成功: {success_count} 张) print(f失败: {len(results) - success_count} 张) print(f总耗时: {total_time:.2f} 秒) if success_count 0: print(f平均每张: {total_time/success_count:.2f} 秒) print(f结果保存在: {args.output} 目录) print(*50) if __name__ __main__: main()7. 总结与最佳实践通过上面的详细讲解和代码示例相信你已经掌握了如何使用requests库远程调用Qwen2.5-VL-Chord视觉定位模型的API。让我们最后总结一下关键要点和最佳实践。7.1 关键要点回顾API调用基础使用requests.post发送JSON数据包含图片的base64编码和文本提示错误处理一定要添加超时设置和异常捕获确保程序稳定性批量处理使用线程池并发处理大幅提高效率结果解析正确解析返回的边界框坐标转换为可用的数据结构可视化验证在本地图片上绘制边界框直观验证识别结果7.2 最佳实践建议性能优化方面根据服务器性能调整并发数一般3-5个线程比较合适对于大量图片可以先压缩到合适尺寸再发送减少传输数据量实现断点续传功能避免网络中断导致全部重来错误处理方面添加重试机制对于临时性错误自动重试记录详细的日志方便排查问题实现熔断机制当错误率过高时暂停调用代码质量方面使用面向对象设计提高代码复用性添加类型注解提高代码可读性编写单元测试确保核心功能稳定7.3 下一步学习建议如果你已经掌握了基础API调用可以进一步学习异步编程使用asyncio和aiohttp实现真正的异步调用分布式处理将任务分发到多台服务器处理海量图片模型微调针对特定场景微调模型提高识别准确率系统集成将视觉定位能力集成到更大的系统中7.4 常见问题快速解决问题API响应很慢怎么办检查网络连接质量减少并发数避免服务器过载压缩图片尺寸减少数据传输量问题识别准确率不高怎么办优化提示词使用更精确的描述确保图片质量避免模糊或过暗对于重要场景可以人工复核或使用多模型投票问题如何处理视频中的视觉定位将视频按帧拆分为图片序列对每张图片调用API将结果按时间顺序整合形成轨迹获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。