Ostrakon-VL-8B代码实例：Python调用vLLM API实现图片base64编码+prompt构造

1. 从界面到代码：为什么需要自己调用API？

如果你用过Chainlit前端和Ostrakon-VL-8B模型对话，可能会觉得挺方便——上传图片、输入问题、点击发送，答案就出来了。但实际工作中，我们往往需要把这种能力集成到自己的系统里，而不是每次都打开一个网页界面。

想象一下这些场景：

• 你的电商系统需要自动分析用户上传的商品图片，生成商品描述
• 你的客服系统需要识别用户发来的问题截图，自动给出解决方案
• 你的内容审核平台需要批量处理图片，检查是否符合规范

这些场景下，你不可能让员工一个个去网页上操作。你需要的是程序化的调用方式——用代码直接和模型对话。

这就是今天要讲的内容：如何用Python代码调用vLLM部署的Ostrakon-VL-8B模型，实现图片识别和分析。我会带你一步步完成从图片准备到API调用的完整流程，让你真正掌握这个多模态模型的编程接口。

2. 准备工作：理解核心概念

在开始写代码之前，我们先搞清楚几个关键概念，这样后面写代码时你才知道每一步在做什么。

2.1 什么是vLLM API？

vLLM是一个高性能的推理引擎，专门为大语言模型设计。当你用vLLM部署了Ostrakon-VL-8B模型后，它会启动一个HTTP服务，这个服务提供了标准的API接口。

简单来说，vLLM API就像是一个"翻译官"：

• 你发送HTTP请求（包含图片和问题）
• vLLM接收请求，把数据传给Ostrakon-VL模型
• 模型处理完后，vLLM把结果包装成HTTP响应返回给你

整个过程和你用浏览器访问Chainlit前端本质上是一样的，只是现在换成了程序自动完成。

2.2 图片为什么要base64编码？

你可能会有疑问：图片不就是文件吗，直接传文件不行吗？

在实际的HTTP API调用中，直接传文件确实可以，但base64编码有几个优势：

1. 简化传输：base64把二进制图片数据转换成纯文本字符串，这样整个请求体都是纯文本的JSON格式，处理起来更统一
2. 避免文件路径问题：如果你的代码和图片不在同一台服务器上，base64编码可以避免复杂的文件传输问题
3. 便于调试：base64字符串可以直接复制粘贴，调试时更方便查看

base64编码的原理很简单：把图片的二进制数据每3个字节（24位）分成一组，然后把这24位数据重新划分为4个6位的片段，每个6位片段对应一个base64字符。最终得到的就是一串由A-Z、a-z、0-9、+、/组成的字符串。

2.3 prompt构造：如何让模型理解你的意图？

多模态模型和纯文本模型不同，它需要同时处理图片和文字。所以你的prompt（提示词）需要告诉模型两件事：

1. 这里有一张图片
2. 关于这张图片，我想问什么问题

Ostrakon-VL模型使用特定的格式来标记图片位置，通常是在prompt中用特殊标记（如<image>）表示图片插入点。

3. 完整代码实现：一步步构建API调用

现在我们来写实际的代码。我会把整个过程拆解成几个步骤，每个步骤都有详细的解释和代码。

3.1 第一步：安装必要的Python库

首先确保你的Python环境已经安装了必要的库。打开终端或命令行，执行：

pip install requests pillow

• requests：用于发送HTTP请求到vLLM API
• pillow：Python的图像处理库，用于打开和操作图片文件

如果你已经部署了Ostrakon-VL-8B模型，这些库应该都已经安装了。如果没有，执行上面的命令安装即可。

3.2 第二步：准备图片并转换为base64

我们先写一个函数来处理图片。这个函数要做三件事：

1. 打开图片文件
2. 检查图片格式（确保是模型支持的格式）
3. 转换为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, format='JPEG')
            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)} 字符")

这个函数有几个关键点需要注意：

1. 图片格式转换：我们统一将图片保存为JPEG格式，因为这是最通用的格式，几乎所有模型都支持
2. 模式检查：有些图片可能是RGBA（带透明度）或其他模式，我们统一转换为RGB模式
3. 错误处理：添加了基本的错误处理，避免程序因为图片问题而崩溃

3.3 第三步：构造完整的prompt

Ostrakon-VL模型需要特定的prompt格式来识别图片。根据模型文档，通常的格式是在prompt中插入<image>标记来表示图片位置。

def construct_prompt(image_base64, question):
    """
    构造包含图片和问题的完整prompt
    
    参数:
        image_base64: base64编码的图片字符串
        question: 要问的问题
        
    返回:
        构造好的prompt字典
    """
    # 这是Ostrakon-VL模型期望的格式
    # <image>标记表示图片插入的位置
    prompt_text = f"<image>\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(f"Prompt文本：{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_url="http://localhost:8000/v1/completions", timeout=60):
    """
    调用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,
            headers=headers,
            data=json.dumps(prompt_data),
            timeout=timeout
        )
        
        # 计算请求耗时
        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, indent=2, ensure_ascii=False))

这个函数有几个重要的错误处理：

1. 超时处理：模型推理可能需要较长时间，我们设置了60秒超时
2. 连接错误：如果vLLM服务没有启动，或者端口被占用，会捕获连接错误
3. 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_url="http://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, format='JPEG', quality=95)
                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_tokens=512, temperature=0.1):
        """
        构造API请求数据
        
        参数:
            image_base64: base64编码的图片
            question: 问题文本
            max_tokens: 最大生成token数
            temperature: 温度参数
            
        返回:
            构造好的请求数据字典
        """
        # Ostrakon-VL模型的prompt格式
        prompt_text = f"<image>\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, timeout=60):
        """
        向图片提问的完整流程
        
        参数:
            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,
                headers=self.headers,
                data=json.dumps(request_data),
                timeout=timeout
            )
            
            elapsed_time = time.time() - start_time
            
            if response.status_code == 200:
                result = response.json()
                print(f"API调用成功！耗时：{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(f"Token使用情况：")
                        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(f"API调用失败，状态码：{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_url="http://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}")

这个完整的类封装了所有功能，你可以直接复制使用。主要特点：

1. 模块化设计：每个功能都有独立的方法，方便维护和扩展
2. 完整的错误处理：每个步骤都有错误检查
3. 详细的日志：打印每个步骤的状态，方便调试
4. 易于使用：只需要调用ask_image_question方法即可

5. 实际应用：批量处理图片

在实际工作中，我们经常需要批量处理多张图片。下面是一个批量处理的示例：

import os
from concurrent.futures import ThreadPoolExecutor, as_completed

class BatchImageProcessor:
    """
    批量图片处理器
    可以同时处理多张图片的问答
    """
    
    def __init__(self, client, max_workers=3):
        """
        初始化批量处理器
        
        参数:
            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_workers=self.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_workers=2)
    
    # 准备要处理的图片和问题
    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("没有找到可处理的图片文件")

这个批量处理器有几个实用功能：

1. 并发处理：使用线程池同时处理多张图片，提高效率
2. 进度显示：实时显示处理进度和成功/失败统计
3. 错误隔离：单张图片处理失败不会影响其他图片
4. 结果汇总：最后提供详细的处理结果报告

6. 常见问题与解决方案

在实际使用中，你可能会遇到一些问题。这里整理了一些常见问题和解决方法。

6.1 图片处理相关的问题

问题1：图片太大，base64编码后字符串太长

def optimize_image_size(image_path, max_size_kb=500):
    """
    优化图片大小，避免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, format='JPEG', quality=quality, optimize=True)
            
            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, mask=img.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', quality=95)
                print(f"已转换并保存为：{temp_path}")
                return temp_path
            else:
                return image_path
                
    except Exception as e:
        print(f"检查图片格式时出错：{e}")
        return None

6.2 API调用相关的问题

问题3：连接超时或失败

def test_api_connection(api_url, timeout=10):
    """
    测试API连接是否正常
    
    参数:
        api_url: API地址
        timeout: 测试超时时间
        
    返回:
        True如果连接正常，False否则
    """
    try:
        # 先尝试简单的HTTP请求
        response = requests.get(api_url.replace('/v1/completions', ''), timeout=timeout)
        if response.status_code == 200:
            print("API服务可访问")
            return True
        else:
            print(f"API服务返回异常状态码：{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(
        total=3,  # 总重试次数
        backoff_factor=1,  # 重试间隔
        status_forcelist=[429, 500, 502, 503, 504],  # 需要重试的状态码
    )
    
    # 配置适配器
    adapter = HTTPAdapter(
        max_retries=retry_strategy,
        pool_connections=10,  # 连接池大小
        pool_maxsize=10
    )
    
    session.mount("http://", adapter)
    session.mount("https://", adapter)
    
    return session

# 在客户端中使用
class OptimizedOstrakonVLClient(OstrakonVLClient):
    def __init__(self, api_url="http://localhost:8000/v1/completions"):
        super().__init__(api_url)
        self.session = create_http_session()
    
    def ask_image_question(self, image_path, question, timeout=60):
        # 使用session代替requests.post
        response = self.session.post(
            self.api_url,
            headers=self.headers,
            data=json.dumps(request_data),
            timeout=timeout
        )
        # ... 其余代码相同

建议2：缓存base64编码结果

import hashlib
from functools import lru_cache

class CachedImageProcessor:
    def __init__(self):
        self.cache = {}
    
    @lru_cache(maxsize=100)
    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_str

7. 总结

通过本文的讲解，你应该已经掌握了如何使用Python调用vLLM部署的Ostrakon-VL-8B模型进行图片识别和分析。让我们回顾一下关键点：

7.1 核心步骤总结

1. 图片准备与编码：使用PIL库打开图片，转换为RGB模式，然后进行base64编码
2. prompt构造：按照<image>\n问题的格式构造prompt，设置合适的生成参数
3. API调用：通过HTTP POST请求发送数据到vLLM服务，处理响应和错误
4. 结果解析：从JSON响应中提取模型生成的文本内容

7.2 实际应用建议

在实际项目中，我建议：

1. 错误处理要完善：网络请求、图片处理、API响应都可能出错，要有完整的错误处理机制
2. 性能要考虑：对于批量处理，使用并发和缓存可以显著提高效率
3. 代码要模块化：像我们示例中的类设计，把不同功能分开，方便维护和测试
4. 日志要详细：记录关键步骤和耗时，方便排查问题

7.3 扩展思路

掌握了基础调用后，你可以进一步扩展：

• 集成到Web应用：使用Flask或FastAPI创建REST API，让其他系统可以调用
• 添加业务逻辑：根据模型返回的结果，执行特定的业务操作
• 结合其他模型：将Ostrakon-VL与其他AI模型结合，实现更复杂的功能
• 自动化工作流：定时处理图片，自动生成报告或触发后续操作

7.4 最后提醒

记住，代码只是工具，真正的价值在于解决实际问题。Ostrakon-VL-8B作为一个专门针对零售和食品服务场景优化的多模态模型，在商品识别、场景分析、合规检查等方面有着独特的优势。结合合适的业务场景，这个技术可以为你创造真正的价值。

现在，你可以开始尝试用这些代码解决自己的实际问题了。如果在使用过程中遇到问题，或者有更好的实现思路，欢迎继续探索和实践。

---

获取更多AI镜像

想探索更多AI镜像和应用场景？访问 CSDN星图镜像广场，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。