Ostrakon-VL-8B基础教程：Python调用API实现批量库存盘点自动化

Ostrakon-VL-8B基础教程：Python调用API实现批量库存盘点自动化

1. 引言

如果你是零售或餐饮行业的从业者，每天面对成百上千的商品库存，手动盘点是不是让你头疼不已？拍照、记录、核对、录入系统……一套流程下来，半天时间就没了，还容易出错。

今天我要分享一个能让你彻底告别手动盘点的自动化方案。我们利用Ostrakon-VL-8B这个专门为零售餐饮场景优化的多模态大模型，通过Python脚本批量处理货架照片，自动识别商品、统计数量，还能检查陈列合规性。整个过程完全自动化，效率提升至少10倍。

Ostrakon-VL-8B是基于Qwen3-VL-8B微调的开源模型，它在商品识别、货架分析、价格标签识别等方面表现特别出色。更重要的是，它提供了WebUI和API两种使用方式，我们可以通过API实现批量自动化处理。

这篇文章我会手把手教你如何用Python调用Ostrakon-VL-8B的API，搭建一个完整的库存盘点自动化系统。即使你只有基础的Python知识，跟着步骤走也能轻松搞定。

2. 环境准备与快速部署

2.1 系统要求检查

在开始之前，我们先确认一下你的环境是否符合要求：

• GPU：需要NVIDIA RTX 4090D或同等性能的显卡，显存至少24GB
• 内存：建议32GB以上
• 存储：至少50GB可用空间
• Python：3.10或更高版本
• 操作系统：Ubuntu 20.04/22.04或CentOS 8+

如果你使用的是云服务器，确保选择带有足够GPU资源的实例。本地部署的话，检查一下显卡驱动是否安装正确。

2.2 一键部署Ostrakon-VL-8B

如果你还没有部署Ostrakon-VL-8B，这里提供最简单的部署方法。假设你已经有了满足要求的服务器环境：

# 1. 克隆项目代码
git clone https://github.com/Ostrakon-VL/Ostrakon-VL.git
cd Ostrakon-VL

# 2. 创建Python虚拟环境
python -m venv venv
source venv/bin/activate

# 3. 安装依赖包
pip install torch torchvision torchaudio –index-url https://download.pytorch.org/whl/cu118
pip install -r requirements.txt

# 4. 下载模型权重
# 如果你有HuggingFace账号，可以直接下载
# 或者从镜像站下载预训练好的模型

# 5. 启动WebUI服务
python webui.py –share –port 7860

部署完成后，在浏览器中打开 http://你的服务器IP:7860 就能看到WebUI界面了。这个界面主要用于测试和手动操作，我们的自动化系统将通过API调用。

2.3 安装Python依赖包

我们需要安装几个关键的Python包来构建自动化系统：

pip install requests pillow opencv-python pandas numpy

简单解释一下这些包的作用：

• requests：用于发送HTTP请求调用API
• pillow：处理图片文件
• opencv-python：图片处理和格式转换
• pandas：处理和分析识别结果数据
• numpy：数值计算

3. Python调用API基础

3.1 理解Ostrakon-VL-8B的API接口

Ostrakon-VL-8B提供了两种API调用方式：

我们先从最简单的同步API开始。API的基本地址是 http://你的服务器IP:7860/api/v1/chat/completions，使用POST方法发送请求。

3.2 第一个API调用示例

让我们写一个最简单的Python脚本来测试API是否正常工作：

import requests
import base64
import json

def test_api_connection(server_url):
"""
测试API连接是否正常
"""
# 准备测试图片（这里用base64编码的简单图片）
# 实际使用时，我们会从文件读取图片
test_image_base64 = "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mNkYPhfDwAChwGA60e6kgAAAABJRU5ErkJggg=="

# 构建请求数据
payload = {
"model": "Ostrakon-VL-8B",
"messages": [
{
"role": "user",
"content": [
{
"type": "text",
"text": "请描述这张图片的内容"
},
{
"type": "image_url",
"image_url": {
"url": f"data:image/jpeg;base64,{test_image_base64}"
}
}
]
}
],
"max_tokens": 500
}

# 发送请求
try:
response = requests.post(
f"{server_url}/api/v1/chat/completions",
json=payload,
timeout=30
)

if response.status_code == 200:
result = response.json()
print("API连接成功！")
print(f"模型回复：{result['choices'][0]['message']['content']}")
return True
else:
print(f"API请求失败，状态码：{response.status_code}")
print(f"错误信息：{response.text}")
return False

except Exception as e:
print(f"连接API时发生错误：{str(e)}")
return False

# 使用示例
if __name__ == "__main__":
server_url = "http://localhost:7860" # 修改为你的服务器地址
test_api_connection(server_url)

运行这个脚本，如果看到"API连接成功！"的提示，说明一切正常。接下来我们就可以开始构建真正的库存盘点系统了。

4. 构建库存盘点自动化系统

4.1 系统架构设计

我们的自动化系统包含以下几个核心模块：

库存盘点自动化系统
├── 图片采集模块（从摄像头或手机获取货架照片）
├── 图片预处理模块（调整大小、格式转换）
├── API调用模块（发送图片给Ostrakon-VL-8B）
├── 结果解析模块（提取商品信息和数量）
├── 数据汇总模块（生成盘点报告）
└── 异常检测模块（识别陈列问题）

整个流程是这样的：

4.2 图片预处理函数

在实际应用中，我们拍摄的照片可能大小不一、格式不同。为了让模型识别更准确，我们需要对图片进行预处理：

from PIL import Image
import cv2
import os

def preprocess_image(image_path, output_size=(1024, 1024)):
"""
预处理图片：调整大小、转换格式、增强质量
"""
try:
# 读取图片
if isinstance(image_path, str):
img = Image.open(image_path)
else:
# 如果是numpy数组（来自摄像头）
img = Image.fromarray(image_path)

# 转换为RGB模式（确保颜色正确）
if img.mode != 'RGB':
img = img.convert('RGB')

# 调整大小，保持宽高比
img.thumbnail(output_size, Image.Resampling.LANCZOS)

# 创建新的画布，将图片放在中间
new_img = Image.new('RGB', output_size, (255, 255, 255))
img_width, img_height = img.size
left = (output_size[0] – img_width) // 2
top = (output_size[1] – img_height) // 2
new_img.paste(img, (left, top))

# 保存预处理后的图片
temp_path = "temp_preprocessed.jpg"
new_img.save(temp_path, "JPEG", quality=95)

return temp_path

except Exception as e:
print(f"图片预处理失败：{str(e)}")
return None

def image_to_base64(image_path):
"""
将图片转换为base64编码
"""
try:
with open(image_path, "rb") as image_file:
encoded_string = base64.b64encode(image_file.read()).decode('utf-8')
return encoded_string
except Exception as e:
print(f"图片转base64失败：{str(e)}")
return None

4.3 核心API调用函数

这是整个系统的核心，负责与Ostrakon-VL-8B通信：

import requests
import time
from typing import List, Dict, Optional

class OstrakonVLClient:
def __init__(self, server_url: str = "http://localhost:7860"):
"""
初始化Ostrakon-VL客户端
"""
self.server_url = server_url
self.api_endpoint = f"{server_url}/api/v1/chat/completions"

def analyze_single_image(self, image_path: str, question: str) -> Optional[Dict]:
"""
分析单张图片
"""
try:
# 预处理图片
processed_path = preprocess_image(image_path)
if not processed_path:
return None

# 转换为base64
image_base64 = image_to_base64(processed_path)
if not image_base64:
return None

# 构建请求
payload = {
"model": "Ostrakon-VL-8B",
"messages": [
{
"role": "user",
"content": [
{"type": "text", "text": question},
{
"type": "image_url",
"image_url": {
"url": f"data:image/jpeg;base64,{image_base64}"
}
}
]
}
],
"max_tokens": 1000,
"temperature": 0.1 # 低温度确保结果稳定
}

# 发送请求
start_time = time.time()
response = requests.post(
self.api_endpoint,
json=payload,
timeout=60 # 图片分析可能需要较长时间
)
elapsed_time = time.time() – start_time

if response.status_code == 200:
result = response.json()
# 清理临时文件
if os.path.exists(processed_path):
os.remove(processed_path)

return {
"success": True,
"response": result['choices'][0]['message']['content'],
"processing_time": elapsed_time
}
else:
print(f"API请求失败：{response.status_code}")
print(f"错误信息：{response.text}")
return {
"success": False,
"error": f"HTTP {response.status_code}: {response.text}"
}

except requests.exceptions.Timeout:
return {
"success": False,
"error": "请求超时，请检查网络连接或服务器状态"
}
except Exception as e:
return {
"success": False,
"error": str(e)
}

def analyze_batch_images(self, image_paths: List[str], questions: List[str]) -> List[Dict]:
"""
批量分析多张图片
注意：Ostrakon-VL-8B目前不支持真正的批量API，
这里是通过循环调用单张图片API实现的
"""
results = []

for i, (image_path, question) in enumerate(zip(image_paths, questions)):
print(f"正在处理第 {i+1}/{len(image_paths)} 张图片…")

result = self.analyze_single_image(image_path, question)
if result:
results.append(result)
else:
results.append({
"success": False,
"error": "处理失败",
"image_path": image_path
})

# 避免请求过于频繁
time.sleep(1)

return results

4.4 库存盘点专用问题模板

针对不同的盘点需求，我们准备了一系列优化过的问题模板。这些问题经过测试，能获得最准确的识别结果：

class InventoryQuestions:
"""
库存盘点专用问题模板
"""

@staticmethod
def get_product_identification_question():
"""商品识别问题"""
return """请仔细识别图片中的所有商品，按以下格式回答：
1. 商品名称：[商品具体名称]
2. 品牌：[品牌名称，如果可见]
3. 数量：[估计数量]
4. 位置描述：[在货架上的位置]
5. 包装规格：[如500ml、1kg等]

请列出所有可见商品，每个商品单独一行。"""

@staticmethod
def get_shelf_analysis_question():
"""货架分析问题"""
return """请分析这个货架的陈列情况：
1. 货架类型：[如饮料架、零食架、冷藏柜等]
2. 陈列层数：[共几层]
3. 每层商品种类：[每层主要是什么商品]
4. 陈列整齐度：[1-10分，10分为最整齐]
5. 需要补货的商品：[哪些商品数量明显不足]
6. 陈列问题：[如商品倒置、标签不清等]"""

@staticmethod
def get_price_check_question():
"""价格标签检查问题"""
return """请检查图片中的价格标签：
1. 可见的价格标签数量：[个]
2. 标签清晰度：[清晰/模糊/不可读]
3. 价格信息是否完整：[是/否]
4. 促销信息：[如有，请描述]
5. 问题标签：[哪些标签有问题，具体问题是什么]"""

@staticmethod
def get_compliance_check_question():
"""合规检查问题"""
return """请检查以下合规项目：
1. 过期商品：[如有，请描述]
2. 破损包装：[如有，请描述]
3. 卫生状况：[清洁/一般/脏乱]
4. 安全通道：[是否畅通]
5. 消防设施：[是否被遮挡]
6. 其他违规项：[如有，请描述]"""

@staticmethod
def get_inventory_count_question(product_name=None):
"""库存数量统计问题"""
if product_name:
return f"""请统计图片中'{product_name}'商品的数量。
注意：请仔细数清楚，包括部分被遮挡的商品。
如果无法确定具体数量，请给出估计范围。"""
else:
return """请统计图片中所有商品的总数量。
按商品种类分别统计，格式为：商品名称: 数量"""

5. 完整库存盘点系统实现

5.1 主程序：自动化盘点流程

现在我们把所有模块组合起来，创建一个完整的库存盘点系统：

import os
import json
import pandas as pd
from datetime import datetime
from pathlib import Path

class InventoryAutomationSystem:
"""
库存盘点自动化系统
"""

def __init__(self, server_url: str, output_dir: str = "./inventory_reports"):
self.client = OstrakonVLClient(server_url)
self.questions = InventoryQuestions()
self.output_dir = output_dir

# 创建输出目录
os.makedirs(output_dir, exist_ok=True)

def process_store_section(self, section_name: str, image_folder: str):
"""
处理一个店铺区域的库存盘点
"""
print(f"\\n{'='*50}")
print(f"开始处理区域：{section_name}")
print(f"{'='*50}")

# 获取所有图片文件
image_extensions = ['.jpg', '.jpeg', '.png', '.bmp', '.webp']
image_files = []

for ext in image_extensions:
image_files.extend(Path(image_folder).glob(f"*{ext}"))
image_files.extend(Path(image_folder).glob(f"*{ext.upper()}"))

if not image_files:
print(f"在 {image_folder} 中未找到图片文件")
return None

print(f"找到 {len(image_files)} 张图片")

# 为每张图片准备问题
questions = []
for img_path in image_files:
# 根据文件名判断问题类型
img_name = img_path.name.lower()

if 'price' in img_name or '标签' in img_name:
questions.append(self.questions.get_price_check_question())
elif 'shelf' in img_name or '货架' in img_name:
questions.append(self.questions.get_shelf_analysis_question())
elif 'compliance' in img_name or '合规' in img_name:
questions.append(self.questions.get_compliance_check_question())
else:
questions.append(self.questions.get_product_identification_question())

# 批量分析图片
results = self.client.analyze_batch_images(
[str(img) for img in image_files],
questions
)

# 解析结果
inventory_data = self._parse_results(results, image_files)

# 生成报告
report = self._generate_report(section_name, inventory_data)

# 保存结果
self._save_results(section_name, inventory_data, report)

return report

def _parse_results(self, results, image_files):
"""
解析API返回的结果
"""
inventory_data = []

for i, (result, img_path) in enumerate(zip(results, image_files)):
if result.get('success'):
response_text = result['response']

# 这里可以根据实际返回格式进行解析
# 示例：简单的关键词提取
item_data = {
'image_file': str(img_path),
'processing_time': result.get('processing_time', 0),
'raw_response': response_text,
'product_count': self._extract_product_count(response_text),
'has_issues': self._check_for_issues(response_text),
'timestamp': datetime.now().strftime('%Y-%m-%d %H:%M:%S')
}

inventory_data.append(item_data)

print(f"图片 {i+1} 分析完成，耗时 {result.get('processing_time', 0):.2f}秒")
else:
print(f"图片 {i+1} 分析失败：{result.get('error', '未知错误')}")

return inventory_data

def _extract_product_count(self, text: str) -> int:
"""
从响应文本中提取商品数量
这是一个简单的示例，实际应用中可能需要更复杂的解析逻辑
"""
import re

# 查找数字模式
patterns = [
r'数量[：:]\\s*(\\d+)',
r'共\\s*(\\d+)\\s*个',
r'总计[：:]\\s*(\\d+)',
r'(\\d+)\\s*件商品'
]

for pattern in patterns:
match = re.search(pattern, text)
if match:
try:
return int(match.group(1))
except:
continue

# 如果没有找到明确的数量，尝试统计提到的商品种类
lines = text.split('\\n')
product_lines = [line for line in lines if '商品' in line or '产品' in line]
return len(product_lines)

def _check_for_issues(self, text: str) -> bool:
"""
检查是否有问题或异常
"""
issue_keywords = ['问题', '异常', '不足', '缺少', '破损', '过期', '脏乱', '遮挡']

for keyword in issue_keywords:
if keyword in text:
return True
return False

def _generate_report(self, section_name: str, inventory_data: list) -> dict:
"""
生成盘点报告
"""
if not inventory_data:
return {"error": "没有有效数据"}

# 计算统计信息
total_images = len(inventory_data)
successful_images = len([d for d in inventory_data if d.get('success', True)])
total_products = sum([d.get('product_count', 0) for d in inventory_data])
images_with_issues = len([d for d in inventory_data if d.get('has_issues', False)])

avg_processing_time = sum([d.get('processing_time', 0) for d in inventory_data]) / total_images

report = {
'section_name': section_name,
'report_date': datetime.now().strftime('%Y-%m-%d'),
'report_time': datetime.now().strftime('%H:%M:%S'),
'total_images_processed': total_images,
'successful_images': successful_images,
'success_rate': f"{(successful_images/total_images*100):.1f}%" if total_images > 0 else "0%",
'total_products_identified': total_products,
'images_with_issues': images_with_issues,
'average_processing_time': f"{avg_processing_time:.2f}秒",
'detailed_data': inventory_data
}

return report

def _save_results(self, section_name: str, inventory_data: list, report: dict):
"""
保存结果到文件
"""
# 生成文件名
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
base_filename = f"{section_name}_{timestamp}"

# 保存原始数据为JSON
json_path = os.path.join(self.output_dir, f"{base_filename}_raw.json")
with open(json_path, 'w', encoding='utf-8') as f:
json.dump({
'inventory_data': inventory_data,
'report': report
}, f, ensure_ascii=False, indent=2)

# 保存报告为CSV（便于Excel打开）
csv_data = []
for item in inventory_data:
csv_data.append({
'图片文件': item['image_file'],
'处理时间(秒)': f"{item.get('processing_time', 0):.2f}",
'商品数量': item.get('product_count', 0),
'存在问题': '是' if item.get('has_issues', False) else '否',
'分析时间': item.get('timestamp', '')
})

if csv_data:
df = pd.DataFrame(csv_data)
csv_path = os.path.join(self.output_dir, f"{base_filename}_report.csv")
df.to_csv(csv_path, index=False, encoding='utf-8-sig')

# 保存汇总报告
summary_path = os.path.join(self.output_dir, f"{base_filename}_summary.txt")
with open(summary_path, 'w', encoding='utf-8') as f:
f.write(f"库存盘点报告 – {section_name}\\n")
f.write(f"生成时间：{report['report_date']} {report['report_time']}\\n")
f.write("="*50 + "\\n\\n")
f.write(f"处理图片总数：{report['total_images_processed']}\\n")
f.write(f"成功分析图片：{report['successful_images']}\\n")
f.write(f"分析成功率：{report['success_rate']}\\n")
f.write(f"识别商品总数：{report['total_products_identified']}\\n")
f.write(f"存在问题图片：{report['images_with_issues']}\\n")
f.write(f"平均处理时间：{report['average_processing_time']}\\n\\n")

if report['images_with_issues'] > 0:
f.write("存在问题详情：\\n")
for item in inventory_data:
if item.get('has_issues', False):
f.write(f"- {item['image_file']}\\n")

print(f"\\n报告已保存到：")
print(f"原始数据：{json_path}")
print(f"CSV报告：{csv_path}")
print(f"汇总报告：{summary_path}")

# 使用示例
if __name__ == "__main__":
# 初始化系统
system = InventoryAutomationSystem(
server_url="http://localhost:7860", # 修改为你的服务器地址
output_dir="./inventory_reports"
)

# 处理一个店铺区域
# 假设图片存放在 ./store_section1 目录下
report = system.process_store_section(
section_name="饮料区",
image_folder="./store_section1"
)

if report:
print("\\n盘点完成！")
print(f"区域：{report['section_name']}")
print(f"识别商品总数：{report['total_products_identified']}")
print(f"分析成功率：{report['success_rate']}")

5.2 批量处理多个区域

如果你的店铺有多个区域需要盘点，可以创建一个批量处理脚本：

def batch_process_store_sections(store_layout):
"""
批量处理整个店铺的所有区域
"""
system = InventoryAutomationSystem(
server_url="http://localhost:7860",
output_dir="./full_store_inventory"
)

all_reports = {}

for section_name, image_folder in store_layout.items():
print(f"\\n开始处理：{section_name}")

report = system.process_store_section(section_name, image_folder)

if report:
all_reports[section_name] = report

# 短暂暂停，避免服务器压力过大
import time
time.sleep(2)

# 生成总报告
generate_summary_report(all_reports)

return all_reports

def generate_summary_report(all_reports):
"""
生成店铺总盘点报告
"""
total_products = 0
total_images = 0
total_issues = 0

summary_lines = []
summary_lines.append("="*60)
summary_lines.append("店铺库存盘点总报告")
summary_lines.append(f"生成时间：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
summary_lines.append("="*60)
summary_lines.append("")

for section_name, report in all_reports.items():
total_products += report.get('total_products_identified', 0)
total_images += report.get('total_images_processed', 0)
total_issues += report.get('images_with_issues', 0)

summary_lines.append(f"【{section_name}】")
summary_lines.append(f" 识别商品：{report.get('total_products_identified', 0)}")
summary_lines.append(f" 处理图片：{report.get('successful_images', 0)}/{report.get('total_images_processed', 0)}")
summary_lines.append(f" 成功率：{report.get('success_rate', '0%')}")
summary_lines.append(f" 存在问题：{report.get('images_with_issues', 0)}张")
summary_lines.append("")

summary_lines.append("="*60)
summary_lines.append("店铺总计")
summary_lines.append("="*60)
summary_lines.append(f"总识别商品数：{total_products}")
summary_lines.append(f"总处理图片数：{total_images}")
summary_lines.append(f"总问题图片数：{total_issues}")

# 保存总报告
summary_path = "./full_store_inventory/summary_report.txt"
with open(summary_path, 'w', encoding='utf-8') as f:
f.write('\\n'.join(summary_lines))

print(f"\\n总报告已保存到：{summary_path}")

# 打印到控制台
print('\\n'.join(summary_lines))

# 定义店铺区域布局
store_layout = {
"饮料区": "./images/beverage_section",
"零食区": "./images/snack_section",
"冷藏柜": "./images/refrigerator",
"收银台": "./images/cashier_area"
}

# 执行批量处理
batch_process_store_sections(store_layout)

6. 实用技巧与优化建议

6.1 提升识别准确率的技巧

在实际使用中，你可能发现某些商品的识别准确率不够高。这里有几个实用技巧：

def optimize_photo_quality(image_path):
"""
优化照片质量以提高识别准确率
"""
# 1. 确保光线充足，避免反光和阴影
# 2. 从正面拍摄，避免角度倾斜
# 3. 确保商品标签清晰可见
# 4. 避免商品重叠遮挡
# 5. 每张照片只包含一个货架区域

tips = """
拍摄建议：
1. 距离货架1-2米，保持水平
2. 确保所有商品标签朝外
3. 避免玻璃反光（冷藏柜）
4. 分区域拍摄，不要试图一张照片拍整个货架
5. 对焦在商品上，而不是背景
"""
return tips

def get_optimized_question(image_context):
"""
根据图片内容优化问题
"""
# 如果是饮料区，问更具体的问题
if "beverage" in image_context or "饮料" in image_context:
return """请识别图片中的饮料商品：
1. 按品牌分类（可口可乐、百事、康师傅等）
2. 统计每个品牌的数量
3. 检查是否有过期商品
4. 价格标签是否清晰可见"""

# 如果是生鲜区，关注新鲜度
elif "fresh" in image_context or "生鲜" in image_context:
return """请检查生鲜商品：
1. 商品名称和种类
2. 新鲜度评估（新鲜/一般/不新鲜）
3. 包装完整性
4. 陈列是否符合卫生标准"""

# 默认问题
else:
return InventoryQuestions.get_product_identification_question()

6.2 性能优化建议

当处理大量图片时，性能可能成为问题。这里有几个优化建议：

import concurrent.futures
import threading

class ConcurrentOstrakonClient(OstrakonVLClient):
"""
支持并发处理的客户端
"""

def __init__(self, server_url: str, max_workers: int = 3):
super().__init__(server_url)
self.max_workers = max_workers
self._lock = threading.Lock()

def analyze_images_concurrently(self, image_paths: List[str], questions: List[str]):
"""
并发分析多张图片
"""
results = []

with concurrent.futures.ThreadPoolExecutor(max_workers=self.max_workers) as executor:
# 提交所有任务
future_to_image = {
executor.submit(self.analyze_single_image, img_path, question): (img_path, question)
for img_path, question in zip(image_paths, questions)
}

# 收集结果
for future in concurrent.futures.as_completed(future_to_image):
img_path, question = future_to_image[future]
try:
result = future.result(timeout=120) # 2分钟超时
results.append(result)
print(f"完成：{img_path}")
except Exception as e:
print(f"处理失败 {img_path}: {str(e)}")
results.append({
"success": False,
"error": str(e),
"image_path": img_path
})

return results

import hashlib
import pickle
from functools import lru_cache

class CachedOstrakonClient(OstrakonVLClient):
"""
带缓存功能的客户端，避免重复分析相同图片
"""

def __init__(self, server_url: str, cache_dir: str = "./cache"):
super().__init__(server_url)
self.cache_dir = cache_dir
os.makedirs(cache_dir, exist_ok=True)

def analyze_single_image(self, image_path: str, question: str) -> Optional[Dict]:
"""
带缓存的图片分析
"""
# 生成缓存键
cache_key = self._generate_cache_key(image_path, question)
cache_file = os.path.join(self.cache_dir, f"{cache_key}.pkl")

# 检查缓存
if os.path.exists(cache_file):
try:
with open(cache_file, 'rb') as f:
cached_result = pickle.load(f)
print(f"使用缓存结果：{image_path}")
return cached_result
except:
pass

# 调用父类方法分析
result = super().analyze_single_image(image_path, question)

# 保存到缓存
if result and result.get('success'):
try:
with open(cache_file, 'wb') as f:
pickle.dump(result, f)
except:
pass

return result

def _generate_cache_key(self, image_path: str, question: str) -> str:
"""
生成缓存键
"""
# 使用文件MD5和问题摘要作为缓存键
with open(image_path, 'rb') as f:
file_hash = hashlib.md5(f.read()).hexdigest()

question_hash = hashlib.md5(question.encode()).hexdigest()

return f"{file_hash}_{question_hash}"

6.3 错误处理与重试机制

在实际生产环境中，网络波动或服务器问题可能导致请求失败。我们需要一个健壮的错误处理机制：

class RobustOstrakonClient(OstrakonVLClient):
"""
带重试机制的客户端
"""

def __init__(self, server_url: str, max_retries: int = 3, retry_delay: int = 5):
super().__init__(server_url)
self.max_retries = max_retries
self.retry_delay = retry_delay

def analyze_single_image_with_retry(self, image_path: str, question: str) -> Optional[Dict]:
"""
带重试机制的图片分析
"""
for attempt in range(self.max_retries):
try:
result = self.analyze_single_image(image_path, question)

if result and result.get('success'):
return result
elif attempt < self.max_retries – 1:
print(f"第{attempt + 1}次尝试失败，{self.retry_delay}秒后重试…")
time.sleep(self.retry_delay)
else:
return result

except Exception as e:
print(f"第{attempt + 1}次尝试异常：{str(e)}")
if attempt < self.max_retries – 1:
time.sleep(self.retry_delay)
else:
return {
"success": False,
"error": f"所有{self.max_retries}次尝试都失败：{str(e)}"
}

return None

def analyze_batch_with_retry(self, image_paths: List[str], questions: List[str]) -> List[Dict]:
"""
批量分析，每个请求都有独立的重试机制
"""
results = []

for i, (image_path, question) in enumerate(zip(image_paths, questions)):
print(f"处理第 {i+1}/{len(image_paths)} 张图片…")

result = self.analyze_single_image_with_retry(image_path, question)
results.append(result if result else {
"success": False,
"error": "处理失败",
"image_path": image_path
})

return results

7. 实际应用案例

7.1 连锁便利店库存盘点

假设你管理着10家连锁便利店，每天需要盘点库存。传统方法需要每个店派专人花2-3小时，现在用我们的自动化系统：

def convenience_store_inventory_chain(store_list):
"""
连锁便利店批量库存盘点
"""
all_store_reports = {}

for store_id, store_info in store_list.items():
print(f"\\n开始处理门店：{store_info['name']} (ID: {store_id})")

# 为每个门店创建独立的输出目录
store_output_dir = f"./inventory_reports/store_{store_id}"
os.makedirs(store_output_dir, exist_ok=True)

# 初始化系统
system = InventoryAutomationSystem(
server_url="http://localhost:7860",
output_dir=store_output_dir
)

# 处理门店的各个区域
store_report = {
'store_id': store_id,
'store_name': store_info['name'],
'sections': {}
}

for section_name, image_folder in store_info['sections'].items():
print(f" 处理区域：{section_name}")

report = system.process_store_section(section_name, image_folder)
if report:
store_report['sections'][section_name] = report

all_store_reports[store_id] = store_report

# 生成门店汇总报告
generate_store_summary(store_report)

# 生成连锁店总报告
generate_chain_summary(all_store_reports)

return all_store_reports

def generate_store_summary(store_report):
"""
生成单个门店的汇总报告
"""
store_id = store_report['store_id']
store_name = store_report['store_name']

total_products = 0
total_issues = 0

for section_name, report in store_report['sections'].items():
total_products += report.get('total_products_identified', 0)
total_issues += report.get('images_with_issues', 0)

summary = f"""
门店盘点报告
================
门店ID：{store_id}
门店名称：{store_name}
盘点时间：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}

统计结果：
– 识别商品总数：{total_products}
– 发现问题数量：{total_issues}
– 处理区域数量：{len(store_report['sections'])}

各区域详情：
"""

for section_name, report in store_report['sections'].items():
summary += f"\\n{section_name}："
summary += f"商品{report.get('total_products_identified', 0)}个"
summary += f"，问题{report.get('images_with_issues', 0)}个"

# 保存报告
report_path = f"./inventory_reports/store_{store_id}/store_summary.txt"
with open(report_path, 'w', encoding='utf-8') as f:
f.write(summary)

print(f"门店报告已保存：{report_path}")

7.2 餐厅后厨库存管理

餐厅后厨的库存管理更加复杂，需要关注食材新鲜度、存储条件等：

def restaurant_kitchen_inventory(kitchen_sections):
"""
餐厅后厨库存管理
"""
system = InventoryAutomationSystem(
server_url="http://localhost:7860",
output_dir="./kitchen_inventory"
)

kitchen_report = {
'inventory_date': datetime.now().strftime('%Y-%m-%d'),
'sections': {},
'alerts': []
}

# 定义后厨专用问题
kitchen_questions = {
'refrigerator': """请检查冷藏柜：
1. 所有食材的名称和数量
2. 食材新鲜度（通过颜色、状态判断）
3. 储存温度是否合适（如有温度计）
4. 是否有过期食材
5. 清洁卫生状况""",

'freezer': """请检查冷冻柜：
1. 冷冻食材的名称和数量
2. 是否有解冻迹象
3. 霜冻厚度（如可见）
4. 储存是否整齐有序""",

'dry_storage': """请检查干料仓库：
1. 所有干货的名称和数量
2. 包装完整性
3. 是否有虫害迹象
4. 储存条件（避光、防潮）""",

'prep_area': """请检查准备区：
1. 正在准备的食材
2. 工具清洁状况
3. 操作台卫生
4. 人员操作规范（如可见）"""
}

for section_name, image_folder in kitchen_sections.items():
print(f"\\n检查后厨区域：{section_name}")

# 获取该区域的问题
question = kitchen_questions.get(section_name,
InventoryQuestions.get_product_identification_question())

# 处理该区域的所有图片
image_files = list(Path(image_folder).glob("*.jpg")) + \\
list(Path(image_folder).glob("*.png"))

if not image_files:
print(f" 未找到图片：{image_folder}")
continue

section_results = []
for img_path in image_files:
result = system.client.analyze_single_image(str(img_path), question)
if result and result.get('success'):
section_results.append(result)

# 检查是否有警报项
response_text = result['response'].lower()
alert_keywords = ['过期', '变质', '不新鲜', '虫害', '脏乱', '违规']

for keyword in alert_keywords:
if keyword in response_text:
kitchen_report['alerts'].append({
'section': section_name,
'image': str(img_path),
'issue': keyword,
'time': datetime.now().strftime('%H:%M:%S')
})
print(f" 发现问题：{section_name} – {keyword}")

kitchen_report['sections'][section_name] = {
'image_count': len(image_files),
'success_count': len(section_results),
'alert_count': len([a for a in kitchen_report['alerts']
if a['section'] == section_name])
}

# 生成后厨库存报告
generate_kitchen_report(kitchen_report)

return kitchen_report

8. 总结

通过这篇文章，我们完整地构建了一个基于Ostrakon-VL-8B的库存盘点自动化系统。让我们回顾一下关键要点：

8.1 核心收获

8.2 实际价值

这个自动化系统带来的实际价值是显而易见的：

• 效率提升：传统手动盘点需要数小时的工作，现在几分钟就能完成
• 准确性提高：减少人为错误，确保库存数据的准确性
• 成本降低：减少人工盘点所需的时间和人力成本
• 实时监控：可以随时进行库存检查，及时发现和解决问题
• 数据积累：自动生成电子化报告，便于数据分析和历史对比

8.3 下一步建议

如果你已经成功部署了这个系统，可以考虑以下几个方向的进一步优化：

8.4 开始行动

现在就开始尝试吧！从最简单的单张图片分析开始，逐步扩展到整个货架、整个店铺。记住几个关键点：

库存盘点不再需要繁琐的手工操作，也不再容易出错。通过Ostrakon-VL-8B和Python自动化，你可以把宝贵的时间用在更有价值的工作上，让技术为你服务。

获取更多AI镜像

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