基础微网站开发可信赖,互联网是做什么工作的,上海在线,临沂网站建设培训班Ostrakon-VL-8B代码实例#xff1a;Python调用vLLM API实现图片base64编码prompt构造 1. 从界面到代码#xff1a;为什么需要自己调用API#xff1f; 如果你用过Chainlit前端和Ostrakon-VL-8B模型对话#xff0c;可能会觉得挺方便——上传图片、输入问题、点击发送#…Ostrakon-VL-8B代码实例Python调用vLLM API实现图片base64编码prompt构造1. 从界面到代码为什么需要自己调用API如果你用过Chainlit前端和Ostrakon-VL-8B模型对话可能会觉得挺方便——上传图片、输入问题、点击发送答案就出来了。但实际工作中我们往往需要把这种能力集成到自己的系统里而不是每次都打开一个网页界面。想象一下这些场景你的电商系统需要自动分析用户上传的商品图片生成商品描述你的客服系统需要识别用户发来的问题截图自动给出解决方案你的内容审核平台需要批量处理图片检查是否符合规范这些场景下你不可能让员工一个个去网页上操作。你需要的是程序化的调用方式——用代码直接和模型对话。这就是今天要讲的内容如何用Python代码调用vLLM部署的Ostrakon-VL-8B模型实现图片识别和分析。我会带你一步步完成从图片准备到API调用的完整流程让你真正掌握这个多模态模型的编程接口。2. 准备工作理解核心概念在开始写代码之前我们先搞清楚几个关键概念这样后面写代码时你才知道每一步在做什么。2.1 什么是vLLM APIvLLM是一个高性能的推理引擎专门为大语言模型设计。当你用vLLM部署了Ostrakon-VL-8B模型后它会启动一个HTTP服务这个服务提供了标准的API接口。简单来说vLLM API就像是一个翻译官你发送HTTP请求包含图片和问题vLLM接收请求把数据传给Ostrakon-VL模型模型处理完后vLLM把结果包装成HTTP响应返回给你整个过程和你用浏览器访问Chainlit前端本质上是一样的只是现在换成了程序自动完成。2.2 图片为什么要base64编码你可能会有疑问图片不就是文件吗直接传文件不行吗在实际的HTTP API调用中直接传文件确实可以但base64编码有几个优势简化传输base64把二进制图片数据转换成纯文本字符串这样整个请求体都是纯文本的JSON格式处理起来更统一避免文件路径问题如果你的代码和图片不在同一台服务器上base64编码可以避免复杂的文件传输问题便于调试base64字符串可以直接复制粘贴调试时更方便查看base64编码的原理很简单把图片的二进制数据每3个字节24位分成一组然后把这24位数据重新划分为4个6位的片段每个6位片段对应一个base64字符。最终得到的就是一串由A-Z、a-z、0-9、、/组成的字符串。2.3 prompt构造如何让模型理解你的意图多模态模型和纯文本模型不同它需要同时处理图片和文字。所以你的prompt提示词需要告诉模型两件事这里有一张图片关于这张图片我想问什么问题Ostrakon-VL模型使用特定的格式来标记图片位置通常是在prompt中用特殊标记如image表示图片插入点。3. 完整代码实现一步步构建API调用现在我们来写实际的代码。我会把整个过程拆解成几个步骤每个步骤都有详细的解释和代码。3.1 第一步安装必要的Python库首先确保你的Python环境已经安装了必要的库。打开终端或命令行执行pip install requests pillowrequests用于发送HTTP请求到vLLM APIpillowPython的图像处理库用于打开和操作图片文件如果你已经部署了Ostrakon-VL-8B模型这些库应该都已经安装了。如果没有执行上面的命令安装即可。3.2 第二步准备图片并转换为base64我们先写一个函数来处理图片。这个函数要做三件事打开图片文件检查图片格式确保是模型支持的格式转换为base64编码字符串import base64 from PIL import Image import io def image_to_base64(image_path): 将图片文件转换为base64编码字符串 参数: image_path: 图片文件路径 返回: base64编码的图片字符串 try: # 打开图片文件 with Image.open(image_path) as img: # 检查图片模式如果不是RGB则转换为RGB if img.mode ! RGB: img img.convert(RGB) # 将图片保存到内存中的字节流 img_byte_arr io.BytesIO() img.save(img_byte_arr, formatJPEG) img_byte_arr img_byte_arr.getvalue() # 转换为base64编码 base64_str base64.b64encode(img_byte_arr).decode(utf-8) return base64_str except FileNotFoundError: print(f错误找不到图片文件 {image_path}) return None except Exception as e: print(f处理图片时出错{e}) return None # 测试一下这个函数 if __name__ __main__: # 替换为你的图片路径 test_image_path test_image.jpg base64_image image_to_base64(test_image_path) if base64_image: print(f图片base64编码的前100个字符{base64_image[:100]}...) print(f编码总长度{len(base64_image)} 字符)这个函数有几个关键点需要注意图片格式转换我们统一将图片保存为JPEG格式因为这是最通用的格式几乎所有模型都支持模式检查有些图片可能是RGBA带透明度或其他模式我们统一转换为RGB模式错误处理添加了基本的错误处理避免程序因为图片问题而崩溃3.3 第三步构造完整的promptOstrakon-VL模型需要特定的prompt格式来识别图片。根据模型文档通常的格式是在prompt中插入image标记来表示图片位置。def construct_prompt(image_base64, question): 构造包含图片和问题的完整prompt 参数: image_base64: base64编码的图片字符串 question: 要问的问题 返回: 构造好的prompt字典 # 这是Ostrakon-VL模型期望的格式 # image标记表示图片插入的位置 prompt_text fimage\n{question} # 构造完整的请求数据 prompt_data { prompt: prompt_text, image: image_base64, max_tokens: 512, # 最大生成token数 temperature: 0.1, # 温度参数控制随机性 top_p: 0.9, # 核采样参数 stop: [], # 停止词可以为空 stream: False # 是否流式输出 } return prompt_data # 测试prompt构造 if __name__ __main__: # 假设我们已经有了base64图片 test_base64 base64_test_string_here # 这里用假数据测试 test_question 图片中的店铺名是什么 prompt_data construct_prompt(test_base64, test_question) print(构造的prompt数据) print(fPrompt文本{prompt_data[prompt]}) print(f图片base64长度{len(prompt_data[image])}) print(f其他参数max_tokens{prompt_data[max_tokens]}, temperature{prompt_data[temperature]})这里有几个参数需要解释max_tokens模型生成的最大token数。token可以理解为词片段512个token大约对应300-400个汉字temperature控制生成随机性的参数。值越小接近0输出越确定、保守值越大接近1输出越随机、有创意top_p核采样参数与temperature配合使用控制生成质量stream是否使用流式输出。如果设为True可以实时看到生成过程但代码会更复杂3.4 第四步发送请求到vLLM API现在我们已经有了图片的base64编码和构造好的prompt接下来就是发送HTTP请求。import requests import json import time def call_ostrakon_vl_api(prompt_data, api_urlhttp://localhost:8000/v1/completions, timeout60): 调用Ostrakon-VL模型的vLLM API 参数: prompt_data: 构造好的prompt数据 api_url: vLLM API的地址 timeout: 请求超时时间秒 返回: 模型返回的结果 # 设置请求头 headers { Content-Type: application/json, Accept: application/json } try: # 记录开始时间 start_time time.time() # 发送POST请求 response requests.post( api_url, headersheaders, datajson.dumps(prompt_data), timeouttimeout ) # 计算请求耗时 elapsed_time time.time() - start_time # 检查响应状态 if response.status_code 200: result response.json() print(f请求成功耗时{elapsed_time:.2f}秒) return result else: print(f请求失败状态码{response.status_code}) print(f错误信息{response.text}) return None except requests.exceptions.Timeout: print(f请求超时{timeout}秒) return None except requests.exceptions.ConnectionError: print(f连接错误请检查API地址{api_url}) print(确保vLLM服务已经启动并且端口8000可以访问) return None except Exception as e: print(f调用API时出错{e}) return None # 测试API调用 if __name__ __main__: # 这里用假数据测试 test_data { prompt: image\n图片中的店铺名是什么, image: base64_test_string, max_tokens: 512, temperature: 0.1, top_p: 0.9, stop: [], stream: False } result call_ostrakon_vl_api(test_data) if result: print(API返回结果) print(json.dumps(result, indent2, ensure_asciiFalse))这个函数有几个重要的错误处理超时处理模型推理可能需要较长时间我们设置了60秒超时连接错误如果vLLM服务没有启动或者端口被占用会捕获连接错误HTTP状态码检查返回的状态码200表示成功其他状态码表示有错误3.5 第五步解析和处理返回结果API调用成功后我们需要从返回的数据中提取出模型生成的文本。def extract_response(api_result): 从API返回结果中提取模型生成的文本 参数: api_result: API返回的完整结果 返回: 提取的文本内容如果提取失败返回None if not api_result: return None try: # vLLM API的标准返回格式 if choices in api_result and len(api_result[choices]) 0: # 获取第一个选择通常只有一个 choice api_result[choices][0] # 提取生成的文本 if text in choice: generated_text choice[text].strip() return generated_text else: print(返回结果中没有找到text字段) return None else: print(返回结果中没有找到choices字段或choices为空) return None except KeyError as e: print(f解析结果时出错缺少字段{e}) return None except Exception as e: print(f解析结果时出错{e}) return None # 测试结果解析 if __name__ __main__: # 模拟API返回结果 mock_result { id: test-123, object: text_completion, created: 1234567890, model: ostrakon-vl-8b, choices: [ { text: 图片中的店铺名是阳光超市位于街道的拐角处。, index: 0, logprobs: None, finish_reason: length } ], usage: { prompt_tokens: 25, total_tokens: 45, completion_tokens: 20 } } extracted_text extract_response(mock_result) if extracted_text: print(f提取的文本{extracted_text})解析结果时需要注意vLLM API的返回格式。不同版本的vLLM可能有细微差别但基本结构是相似的。4. 完整示例一个可运行的端到端程序现在我们把所有步骤组合起来创建一个完整的示例程序。import base64 import json import time import io import requests from PIL import Image class OstrakonVLClient: Ostrakon-VL模型客户端类 封装了图片处理、prompt构造、API调用的完整流程 def __init__(self, api_urlhttp://localhost:8000/v1/completions): 初始化客户端 参数: api_url: vLLM API地址默认为本地8000端口 self.api_url api_url self.headers { Content-Type: application/json, Accept: application/json } def image_to_base64(self, image_path): 将图片转换为base64编码 参数: image_path: 图片文件路径 返回: base64编码字符串失败返回None try: with Image.open(image_path) as img: # 统一转换为RGB模式的JPEG格式 if img.mode ! RGB: img img.convert(RGB) # 保存到内存字节流 img_byte_arr io.BytesIO() img.save(img_byte_arr, formatJPEG, quality95) img_byte_arr img_byte_arr.getvalue() # base64编码 base64_str base64.b64encode(img_byte_arr).decode(utf-8) return base64_str except Exception as e: print(f转换图片失败{e}) return None def construct_request_data(self, image_base64, question, max_tokens512, temperature0.1): 构造API请求数据 参数: image_base64: base64编码的图片 question: 问题文本 max_tokens: 最大生成token数 temperature: 温度参数 返回: 构造好的请求数据字典 # Ostrakon-VL模型的prompt格式 prompt_text fimage\n{question} request_data { prompt: prompt_text, image: image_base64, max_tokens: max_tokens, temperature: temperature, top_p: 0.9, stop: [], stream: False } return request_data def ask_image_question(self, image_path, question, timeout60): 向图片提问的完整流程 参数: image_path: 图片文件路径 question: 问题文本 timeout: 请求超时时间秒 返回: 模型回答的文本失败返回None print(f处理图片{image_path}) print(f问题{question}) print(- * 50) # 1. 转换图片为base64 print(步骤1转换图片为base64编码...) image_base64 self.image_to_base64(image_path) if not image_base64: print(图片转换失败) return None print(f图片base64编码完成长度{len(image_base64)} 字符) # 2. 构造请求数据 print(步骤2构造请求数据...) request_data self.construct_request_data(image_base64, question) # 3. 调用API print(步骤3调用vLLM API...) start_time time.time() try: response requests.post( self.api_url, headersself.headers, datajson.dumps(request_data), timeouttimeout ) elapsed_time time.time() - start_time if response.status_code 200: result response.json() print(fAPI调用成功耗时{elapsed_time:.2f}秒) # 4. 解析结果 print(步骤4解析返回结果...) if choices in result and len(result[choices]) 0: answer result[choices][0][text].strip() # 打印使用统计 if usage in result: usage result[usage] print(fToken使用情况) print(f 输入token{usage.get(prompt_tokens, N/A)}) print(f 生成token{usage.get(completion_tokens, N/A)}) print(f 总token{usage.get(total_tokens, N/A)}) return answer else: print(返回结果格式错误) return None else: print(fAPI调用失败状态码{response.status_code}) print(f错误信息{response.text}) return None except requests.exceptions.Timeout: print(f请求超时{timeout}秒) return None except Exception as e: print(f调用过程中出错{e}) return None # 使用示例 if __name__ __main__: # 创建客户端实例 # 如果你的vLLM服务在其他地址修改这里的api_url client OstrakonVLClient(api_urlhttp://localhost:8000/v1/completions) # 示例1基本使用 print(示例1基本图片问答) image_path shop_image.jpg # 替换为你的图片路径 question 图片中的店铺名是什么 answer client.ask_image_question(image_path, question) if answer: print(f\n模型回答{answer}) else: print(调用失败) print(\n *50 \n) # 示例2不同的问题 print(示例2询问更多细节) question2 描述一下图片中的场景包括店铺外观、周围环境和任何可见的商品。 answer2 client.ask_image_question(image_path, question2) if answer2: print(f\n模型回答{answer2})这个完整的类封装了所有功能你可以直接复制使用。主要特点模块化设计每个功能都有独立的方法方便维护和扩展完整的错误处理每个步骤都有错误检查详细的日志打印每个步骤的状态方便调试易于使用只需要调用ask_image_question方法即可5. 实际应用批量处理图片在实际工作中我们经常需要批量处理多张图片。下面是一个批量处理的示例import os from concurrent.futures import ThreadPoolExecutor, as_completed class BatchImageProcessor: 批量图片处理器 可以同时处理多张图片的问答 def __init__(self, client, max_workers3): 初始化批量处理器 参数: client: OstrakonVLClient实例 max_workers: 最大并发数 self.client client self.max_workers max_workers def process_single_image(self, image_info): 处理单张图片 参数: image_info: 包含图片路径和问题的字典 返回: 处理结果字典 image_path image_info[path] question image_info[question] print(f开始处理{os.path.basename(image_path)}) result { image: image_path, question: question, success: False, answer: None, error: None } try: answer self.client.ask_image_question(image_path, question) if answer: result[success] True result[answer] answer else: result[error] API调用失败 except Exception as e: result[error] str(e) return result def process_batch(self, image_questions): 批量处理多张图片 参数: image_questions: 列表每个元素是包含path和question的字典 返回: 处理结果列表 results [] print(f开始批量处理 {len(image_questions)} 张图片) print(f使用 {self.max_workers} 个并发线程) print(- * 50) # 使用线程池并发处理 with ThreadPoolExecutor(max_workersself.max_workers) as executor: # 提交所有任务 future_to_image { executor.submit(self.process_single_image, iq): iq for iq in image_questions } # 收集结果 for future in as_completed(future_to_image): image_info future_to_image[future] try: result future.result() results.append(result) # 打印进度 success_count sum(1 for r in results if r[success]) total_count len(results) print(f进度{total_count}/{len(image_questions)}成功{success_count}) except Exception as e: print(f处理 {image_info[path]} 时出错{e}) # 统计结果 success_count sum(1 for r in results if r[success]) print(f\n批量处理完成) print(f总计{len(results)}成功{success_count}失败{len(results)-success_count}) return results # 批量处理示例 if __name__ __main__: # 创建客户端 client OstrakonVLClient() # 创建批量处理器 processor BatchImageProcessor(client, max_workers2) # 准备要处理的图片和问题 image_questions [ { path: shop1.jpg, question: 图片中的店铺名是什么主要卖什么商品 }, { path: shop2.jpg, question: 描述店铺的外观和招牌 }, { path: product1.jpg, question: 这是什么商品估计价格是多少 }, { path: product2.jpg, question: 商品包装上有什么重要信息 } ] # 实际使用时需要确保图片文件存在 # 这里先过滤掉不存在的文件 existing_images [] for iq in image_questions: if os.path.exists(iq[path]): existing_images.append(iq) else: print(f警告图片不存在 {iq[path]}) if existing_images: # 执行批量处理 results processor.process_batch(existing_images) # 打印详细结果 print(\n详细结果) for i, result in enumerate(results, 1): print(f\n{i}. 图片{os.path.basename(result[image])}) print(f 问题{result[question]}) if result[success]: print(f 回答{result[answer][:100]}...) # 只显示前100字符 else: print(f 错误{result[error]}) else: print(没有找到可处理的图片文件)这个批量处理器有几个实用功能并发处理使用线程池同时处理多张图片提高效率进度显示实时显示处理进度和成功/失败统计错误隔离单张图片处理失败不会影响其他图片结果汇总最后提供详细的处理结果报告6. 常见问题与解决方案在实际使用中你可能会遇到一些问题。这里整理了一些常见问题和解决方法。6.1 图片处理相关的问题问题1图片太大base64编码后字符串太长def optimize_image_size(image_path, max_size_kb500): 优化图片大小避免base64字符串过长 参数: image_path: 图片路径 max_size_kb: 目标最大大小KB 返回: 优化后的base64字符串 from PIL import Image import io with Image.open(image_path) as img: # 如果图片模式不是RGB先转换 if img.mode ! RGB: img img.convert(RGB) # 如果图片太大先调整尺寸 max_dimension 1024 # 最大边长 if max(img.size) max_dimension: ratio max_dimension / max(img.size) new_size (int(img.size[0] * ratio), int(img.size[1] * ratio)) img img.resize(new_size, Image.Resampling.LANCZOS) # 逐步降低质量直到满足大小要求 quality 95 while quality 10: img_byte_arr io.BytesIO() img.save(img_byte_arr, formatJPEG, qualityquality, optimizeTrue) size_kb len(img_byte_arr.getvalue()) / 1024 if size_kb max_size_kb: break quality - 5 # 转换为base64 base64_str base64.b64encode(img_byte_arr.getvalue()).decode(utf-8) print(f图片优化原始尺寸 {img.size}质量 {quality}%大小 {size_kb:.1f}KB) return base64_str问题2图片格式不支持def check_and_convert_image(image_path): 检查并转换图片格式 参数: image_path: 图片路径 返回: 转换后的图片路径或base64字符串 from PIL import Image import os supported_formats [JPEG, PNG, BMP, GIF, TIFF] try: with Image.open(image_path) as img: format img.format if format not in supported_formats: print(f警告图片格式 {format} 可能不被完全支持) print(正在转换为JPEG格式...) # 转换为JPEG if img.mode in (RGBA, LA, P): # 如果有透明度添加白色背景 background Image.new(RGB, img.size, (255, 255, 255)) if img.mode P: img img.convert(RGBA) background.paste(img, maskimg.split()[-1] if img.mode RGBA else None) img background elif img.mode ! RGB: img img.convert(RGB) # 保存为临时文件 temp_path f{os.path.splitext(image_path)[0]}_converted.jpg img.save(temp_path, JPEG, quality95) print(f已转换并保存为{temp_path}) return temp_path else: return image_path except Exception as e: print(f检查图片格式时出错{e}) return None6.2 API调用相关的问题问题3连接超时或失败def test_api_connection(api_url, timeout10): 测试API连接是否正常 参数: api_url: API地址 timeout: 测试超时时间 返回: True如果连接正常False否则 try: # 先尝试简单的HTTP请求 response requests.get(api_url.replace(/v1/completions, ), timeouttimeout) if response.status_code 200: print(API服务可访问) return True else: print(fAPI服务返回异常状态码{response.status_code}) return False except requests.exceptions.ConnectionError: print(无法连接到API服务请检查) print(1. vLLM服务是否已启动) print(2. 端口号是否正确默认8000) print(3. 防火墙是否阻止了连接) return False except Exception as e: print(f测试连接时出错{e}) return False问题4返回结果解析错误def safe_extract_response(api_result): 安全地提取API响应兼容不同格式 参数: api_result: API返回结果 返回: 提取的文本或错误信息 if not api_result: return 错误API返回结果为空 # 尝试多种可能的响应格式 try: # 格式1标准vLLM格式 if isinstance(api_result, dict) and choices in api_result: choices api_result[choices] if choices and isinstance(choices, list) and len(choices) 0: first_choice choices[0] if isinstance(first_choice, dict) and text in first_choice: return first_choice[text].strip() # 格式2直接包含text字段 if isinstance(api_result, dict) and text in api_result: return api_result[text].strip() # 格式3已经是字符串 if isinstance(api_result, str): return api_result.strip() # 其他格式尝试转换为字符串 return str(api_result).strip() except Exception as e: return f解析响应时出错{e}。原始响应{api_result}6.3 性能优化建议建议1使用连接池import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry def create_http_session(): 创建配置了重试和连接池的HTTP会话 session requests.Session() # 配置重试策略 retry_strategy Retry( total3, # 总重试次数 backoff_factor1, # 重试间隔 status_forcelist[429, 500, 502, 503, 504], # 需要重试的状态码 ) # 配置适配器 adapter HTTPAdapter( max_retriesretry_strategy, pool_connections10, # 连接池大小 pool_maxsize10 ) session.mount(http://, adapter) session.mount(https://, adapter) return session # 在客户端中使用 class OptimizedOstrakonVLClient(OstrakonVLClient): def __init__(self, api_urlhttp://localhost:8000/v1/completions): super().__init__(api_url) self.session create_http_session() def ask_image_question(self, image_path, question, timeout60): # 使用session代替requests.post response self.session.post( self.api_url, headersself.headers, datajson.dumps(request_data), timeouttimeout ) # ... 其余代码相同建议2缓存base64编码结果import hashlib from functools import lru_cache class CachedImageProcessor: def __init__(self): self.cache {} lru_cache(maxsize100) def get_image_hash(self, image_path): 计算图片的哈希值用于缓存键 with open(image_path, rb) as f: return hashlib.md5(f.read()).hexdigest() def image_to_base64_cached(self, image_path): 带缓存的图片转base64 image_hash self.get_image_hash(image_path) if image_hash in self.cache: print(f使用缓存{image_path}) return self.cache[image_hash] # 计算base64 base64_str self.image_to_base64(image_path) # 存入缓存 if base64_str: self.cache[image_hash] base64_str return base64_str7. 总结通过本文的讲解你应该已经掌握了如何使用Python调用vLLM部署的Ostrakon-VL-8B模型进行图片识别和分析。让我们回顾一下关键点7.1 核心步骤总结图片准备与编码使用PIL库打开图片转换为RGB模式然后进行base64编码prompt构造按照image\n问题的格式构造prompt设置合适的生成参数API调用通过HTTP POST请求发送数据到vLLM服务处理响应和错误结果解析从JSON响应中提取模型生成的文本内容7.2 实际应用建议在实际项目中我建议错误处理要完善网络请求、图片处理、API响应都可能出错要有完整的错误处理机制性能要考虑对于批量处理使用并发和缓存可以显著提高效率代码要模块化像我们示例中的类设计把不同功能分开方便维护和测试日志要详细记录关键步骤和耗时方便排查问题7.3 扩展思路掌握了基础调用后你可以进一步扩展集成到Web应用使用Flask或FastAPI创建REST API让其他系统可以调用添加业务逻辑根据模型返回的结果执行特定的业务操作结合其他模型将Ostrakon-VL与其他AI模型结合实现更复杂的功能自动化工作流定时处理图片自动生成报告或触发后续操作7.4 最后提醒记住代码只是工具真正的价值在于解决实际问题。Ostrakon-VL-8B作为一个专门针对零售和食品服务场景优化的多模态模型在商品识别、场景分析、合规检查等方面有着独特的优势。结合合适的业务场景这个技术可以为你创造真正的价值。现在你可以开始尝试用这些代码解决自己的实际问题了。如果在使用过程中遇到问题或者有更好的实现思路欢迎继续探索和实践。获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。