Trong bài viết này, tôi sẽ chia sẻ kinh nghiệm thực chiến khi triển khai multimodal function calling để parse hình ảnh và trích xuất dữ liệu structured. Sau 2 năm làm việc với các hệ thống AI tại HolySheep AI, tôi đã rút ra nhiều bài học quý giá về cách tối ưu hóa pipeline này.
Tại sao cần Multimodal Function Calling?
Khi xây dựng hệ thống OCR thế hệ mới hoặc document processing pipeline, traditional approach có nhiều hạn chế:
- OCR engine riêng biệt cần training và maintain
- Layout analysis phức tạp và dễ sai
- Khó handle mixed content (text + tables + images)
- Chi phí infrastructure cao
Multimodal models như GPT-4o, Claude Sonnet 4.5 và Gemini 2.5 Flash giải quyết triệt để những vấn đề này. Với function calling, bạn có thể trích xuất structured data một cách deterministic, giảm hallucination đáng kể.
Kiến trúc tổng quan
┌─────────────────────────────────────────────────────────────┐
│ MULTIMODAL PIPELINE │
├─────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────┐ ┌──────────────┐ ┌──────────────────┐ │
│ │ Image │───▶│ Preprocess │───▶│ Function Call │ │
│ │ Upload │ │ & Compress │ │ Definition │ │
│ └──────────┘ └──────────────┘ └────────┬─────────┘ │
│ │ │
│ ┌───────────────────────┘ │
│ ▼ │
│ ┌──────────────────┐ │
│ │ Model Engine │ (GPT-4o / Claude / Gemini)│
│ └────────┬─────────┘ │
│ │ │
│ ┌──────────────┼──────────────┐ │
│ ▼ ▼ ▼ │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ Invoice │ │ ID Card │ │ Receipt │ │
│ │ Parser │ │ Parser │ │ Parser │ │
│ └──────────┘ └──────────┘ └──────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘
Setup và Cấu hình HolySheep AI
Trước tiên, bạn cần đăng ký tại đây để lấy API key. HolySheep AI cung cấp:
- Tỷ giá ¥1 = $1 (tiết kiệm 85%+ so với OpenAI)
- Hỗ trợ WeChat/Alipay cho người dùng Trung Quốc
- Latency trung bình <50ms
- Tín dụng miễn phí khi đăng ký
# Cài đặt dependencies
pip install openai httpx pillow python-multipart asyncio aiofiles
Cấu hình environment
import os
from openai import OpenAI
KHÔNG BAO GIỜ dùng api.openai.com
client = OpenAI(
api_key="YOUR_HOLYSHEEP_API_KEY",
base_url="https://api.holysheep.ai/v1" # Chỉ dùng HolySheep AI
)
Giá tham khảo 2026 (USD/MTok)
PRICING = {
"gpt-4.1": 8.0, # GPT-4.1
"claude-sonnet-4.5": 15.0, # Claude Sonnet 4.5
"gemini-2.5-flash": 2.50, # Gemini 2.5 Flash
"deepseek-v3.2": 0.42 # DeepSeek V3.2 - TIẾT KIỆM NHẤT
}
Function Definition cho Image Parsing
Đây là phần quan trọng nhất. Tôi sẽ định nghĩa các function để parse 3 loại document phổ biến: invoice, ID card, và receipt.
import base64
import json
from typing import Optional, List
from pydantic import BaseModel, Field
============== SCHEMA DEFINITIONS ==============
class Address(BaseModel):
street: str = Field(description="Địa chỉ đường phố")
city: str = Field(description="Thành phố")
district: Optional[str] = Field(default=None, description="Quận/Huyện")
country: str = Field(default="Vietnam", description="Quốc gia")
class InvoiceItem(BaseModel):
name: str = Field(description="Tên sản phẩm")
quantity: int = Field(description="Số lượng")
unit_price: float = Field(description="Đơn giá")
total: float = Field(description="Thành tiền")
class InvoiceData(BaseModel):
invoice_number: str = Field(description="Số hóa đơn")
date: str = Field(description="Ngày phát hành")
vendor_name: str = Field(description="Tên nhà cung cấp")
vendor_address: Optional[Address] = Field(default=None)
customer_name: str = Field(description="Tên khách hàng")
customer_address: Optional[Address] = Field(default=None)
items: List[InvoiceItem] = Field(description="Danh sách sản phẩm")
subtotal: float = Field(description="Tổng phụ")
tax: float = Field(description="Thuế")
total: float = Field(description="Tổng cộng")
currency: str = Field(default="VND", description="Đơn vị tiền tệ")
class IDCardData(BaseModel):
full_name: str = Field(description="Họ và tên")
date_of_birth: str = Field(description="Ngày sinh")
gender: str = Field(description="Giới tính")
nationality: str = Field(description="Quốc tịch")
id_number: str = Field(description="Số CMND/CCCD")
issue_date: Optional[str] = Field(default=None, description="Ngày cấp")
expiry_date: Optional[str] = Field(default=None, description="Ngày hết hạn")
address: Optional[str] = Field(default=None, description="Địa chỉ thường trú")
class ReceiptData(BaseModel):
receipt_number: str = Field(description="Số biên nhận")
date: str = Field(description="Ngày giao dịch")
store_name: str = Field(description="Tên cửa hàng")
items: List[InvoiceItem] = Field(description="Danh sách sản phẩm")
total: float = Field(description="Tổng cộng")
payment_method: Optional[str] = Field(default=None, description="Phương thức thanh toán")
============== FUNCTION DEFINITIONS ==============
TOOL_SCHEMA = {
"type": "function",
"function": {
"name": "parse_document",
"description": "Trích xuất thông tin cấu trúc từ hình ảnh document (invoice, ID card, receipt)",
"parameters": {
"type": "object",
"properties": {
"document_type": {
"type": "string",
"enum": ["invoice", "id_card", "receipt"],
"description": "Loại document cần parse"
},
"parsed_data": {
"type": "object",
"description": "Dữ liệu đã trích xuất từ document"
},
"confidence_score": {
"type": "number",
"minimum": 0,
"maximum": 1,
"description": "Độ tin cậy của kết quả (0-1)"
},
"warnings": {
"type": "array",
"items": {"type": "string"},
"description": "Cảnh báo về các trường không đọc được hoặc không chắc chắn"
}
},
"required": ["document_type", "parsed_data", "confidence_score"]
}
}
}
Image Preprocessing và Optimization
Đây là bước tôi thường bỏ qua khi mới bắt đầu, nhưng nó cực kỳ quan trọng cho cost optimization. Image quality直接影响token consumption.
import httpx
import io
from PIL import Image
from PIL.Image import Image as PILImage
import asyncio
from functools import lru_cache
class ImagePreprocessor:
"""Tối ưu hóa hình ảnh trước khi gửi lên API"""
MAX_DIMENSION = 2048 # Kích thước tối đa
COMPRESSION_QUALITY = 85
TARGET_FORMATS = ["png", "webp"]
def __init__(self, max_dimension: int = 2048, quality: int = 85):
self.max_dimension = max_dimension
self.quality = quality
def resize_if_needed(self, img: PILImage) -> PILImage:
"""Resize ảnh nếu vượt kích thước cho phép"""
width, height = img.size
if width <= self.max_dimension and height <= self.max_dimension:
return img
# Tính tỷ lệ scale
if width > height:
new_width = self.max_dimension
new_height = int(height * (self.max_dimension / width))
else:
new_height = self.max_dimension
new_width = int(width * (self.max_dimension / height))
return img.resize((new_width, new_height), Image.Resampling.LANCZOS)
def compress_image(self, img: PILImage, format: str = "PNG") -> bytes:
"""Nén hình ảnh giảm kích thước file"""
buffer = io.BytesIO()
if format.upper() == "JPEG" or format.lower() == "jpg":
# Chuyển sang RGB nếu cần (JPEG không hỗ trợ RGBA)
if img.mode in ("RGBA", "P"):
img = img.convert("RGB")
img.save(buffer, format="JPEG", quality=self.quality, optimize=True)
else:
img.save(buffer, format="PNG", optimize=True)
return buffer.getvalue()
async def preprocess_async(self, image_path: str) -> dict:
"""Preprocess hình ảnh bất đồng bộ"""
loop = asyncio.get_event_loop()
def _process():
img = Image.open(image_path)
# Convert RGBA sang RGB nếu cần
if img.mode == "RGBA":
background = Image.new("RGB", img.size, (255, 255, 255))
background.paste(img, mask=img.split()[-1])
img = background
# Resize
img_resized = self.resize_if_needed(img)
# Compress
image_bytes = self.compress_image(img_resized)
# Base64 encode
base64_image = base64.b64encode(image_bytes).decode("utf-8")
return {
"base64": base64_image,
"size_bytes": len(image_bytes),
"dimensions": img_resized.size,
"mime_type": "image/jpeg"
}
return await loop.run_in_executor(None, _process)
@staticmethod
def estimate_token_cost(image_bytes: int, model: str) -> float:
"""Ước tính chi phí token cho hình ảnh"""
# Ước tính: 1 token ≈ 4 characters, 1 pixel = ~0.75 tokens sau compression
estimated_tokens = int(image_bytes * 0.75)
pricing_per_mtok = PRICING.get(model, 8.0)
cost = (estimated_tokens / 1_000_000) * pricing_per_mtok
return round(cost, 6)
Sử dụng singleton pattern
preprocessor = ImagePreprocessor(max_dimension=2048, quality=85)
Multimodal Parser Engine với Concurrency Control
Đây là core implementation. Tôi đã optimize để handle batch processing hiệu quả với semaphore cho concurrency control.
import asyncio
from typing import Union, List
from dataclasses import dataclass
from datetime import datetime
import time
@dataclass
class ParsingResult:
"""Kết quả parsing với metadata"""
document_type: str
data: Union[InvoiceData, IDCardData, ReceiptData]
confidence: float
warnings: List[str]
latency_ms: float
cost_usd: float
tokens_used: int
model: str
@dataclass
class BatchResult:
"""Kết quả batch processing"""
results: List[ParsingResult]
total_latency_ms: float
total_cost_usd: float
success_count: int
failed_count: int
class MultimodalParser:
"""
Multimodal Parser với Concurrency Control và Cost Optimization
Author: HolySheep AI Engineering Team
"""
SYSTEM_PROMPT = """Bạn là chuyên gia OCR và trích xuất dữ liệu từ document.
Nhiệm vụ của bạn:
1. Xác định loại document (invoice, ID card, receipt)
2. Trích xuất thông tin một cách CHÍNH XÁC
3. Đánh giá confidence score dựa trên chất lượng hình ảnh và độ rõ ràng của text
4. Gọi function parse_document với dữ liệu trích xuất được
QUAN TRỌNG:
- Chỉ trích xuất thông tin bạn NHÌN THẤY rõ ràng
- Nếu không chắc chắn, đánh dấu trong warnings
- Giữ nguyên format số (không làm tròn)
- Trả về JSON hợp lệ
"""
def __init__(
self,
client: OpenAI,
model: str = "gpt-4.1",
max_concurrent: int = 5,
timeout: int = 60
):
self.client = client
self.model = model
self.semaphore = asyncio.Semaphore(max_concurrent)
self.timeout = timeout
self._stats = {"requests": 0, "total_cost": 0.0, "total_tokens": 0}
async def parse_single(
self,
image_path: str,
expected_type: str = "invoice"
) -> ParsingResult:
"""Parse một hình ảnh đơn lẻ với concurrency control"""
async with self.semaphore: # Giới hạn concurrent requests
start_time = time.perf_counter()
try:
# Preprocess image
preprocessed = await preprocessor.preprocess_async(image_path)
# Estimate cost trước
estimated_cost = ImagePreprocessor.estimate_token_cost(
preprocessed["size_bytes"], self.model
)
# Build message với image
messages = [
{
"role": "system",
"content": self.SYSTEM_PROMPT
},
{
"role": "user",
"content": [
{
"type": "text",
"text": f"Hãy parse document này và trích xuất thông tin. Document type: {expected_type}"
},
{
"type": "image_url",
"image_url": {
"url": f"data:image/jpeg;base64,{preprocessed['base64']}",
"detail": "high" # "low", "high", "auto"
}
}
]
}
]
# Gọi API với timeout
response = await asyncio.wait_for(
self._call_api(messages),
timeout=self.timeout
)
# Parse response
result = self._parse_response(response)
latency_ms = (time.perf_counter() - start_time) * 1000
# Update stats
self._stats["requests"] += 1
self._stats["total_cost"] += result.cost_usd
return result
except asyncio.TimeoutError:
raise TimeoutError(f"Parse timeout sau {self.timeout}s")
except Exception as e:
raise RuntimeError(f"Parse failed: {str(e)}")
async def _call_api(self, messages: list) -> dict:
"""Gọi API bất đồng bộ"""
loop = asyncio.get_event_loop()
def _sync_call():
return self.client.chat.completions.create(
model=self.model,
messages=messages,
tools=[TOOL_SCHEMA],
tool_choice="auto",
temperature=0.1 # Low temperature cho deterministic output
)
return await loop.run_in_executor(None, _sync_call)
def _parse_response(self, response) -> ParsingResult:
"""Parse API response thành structured result"""
# Extract usage stats
usage = response.usage
total_tokens = usage.prompt_tokens + usage.completion_tokens
# Tính cost thực tế
cost_per_mtok = PRICING.get(self.model, 8.0)
cost_usd = (total_tokens / 1_000_000) * cost_per_mtok
# Parse function call
tool_calls = response.choices[0].message.tool_calls
if not tool_calls:
raise ValueError("No function call in response")
call = tool_calls[0]
args = json.loads(call.function.arguments)
# Map sang Pydantic models
doc_type = args.get("document_type")
parsed_data = args.get("parsed_data")
if doc_type == "invoice":
data = InvoiceData(**parsed_data)
elif doc_type == "id_card":
data = IDCardData(**parsed_data)
else:
data = ReceiptData(**parsed_data)
return ParsingResult(
document_type=doc_type,
data=data,
confidence=args.get("confidence_score", 0.0),
warnings=args.get("warnings", []),
latency_ms=0, # Sẽ được set bên ngoài
cost_usd=round(cost_usd, 6),
tokens_used=total_tokens,
model=self.model
)
async def parse_batch(
self,
image_paths: List[str],
expected_types: List[str] = None
) -> BatchResult:
"""Parse nhiều hình ảnh song song với rate limiting"""
if expected_types is None:
expected_types = ["invoice"] * len(image_paths)
start_time = time.perf_counter()
# Create tasks
tasks = [
self.parse_single(path, doc_type)
for path, doc_type in zip(image_paths, expected_types)
]
# Execute với gather (giới hạn bởi semaphore)
results = await asyncio.gather(*tasks, return_exceptions=True)
# Filter successful results
successful = [r for r in results if isinstance(r, ParsingResult)]
failed = [r for r in results if isinstance(r, Exception)]
total_latency = (time.perf_counter() - start_time) * 1000
total_cost = sum(r.cost_usd for r in successful)
return BatchResult(
results=successful,
total_latency_ms=round(total_latency, 2),
total_cost_usd=round(total_cost, 4),
success_count=len(successful),
failed_count=len(failed)
)
def get_stats(self) -> dict:
"""Lấy thống kê sử dụng"""
return {
**self._stats,
"avg_cost_per_request":