Bạn đã có từng mắt xích. Bài này ghép chúng thành một agent chạy đầu-cuối: nhận file → phân loại → OCR → trích → kiểm tra → ghi hoặc đẩy duyệt. Điểm mới quan trọng: định tuyến (routing) để mỗi loại tài liệu đi đúng nhánh xử lý.
Kiến trúc điều phối (orchestration)
┌─> [Hóa đơn] -> extract_invoice -> validate_invoice
[Ingest]->[OCR]->[Phân loại]->[Hợp đồng] -> extract_contract -> check_clauses
└─> [Báo cáo] -> extract_report -> validate_report
-> [Phân làn] -> Ghi / Hàng đợi duyệt
Một "nhạc trưởng" (orchestrator) gọi tuần tự các bước và quyết định nhánh dựa trên loại tài liệu.
Bước phân loại tài liệu
Trước khi trích, phải biết đây là gì. Dùng tín hiệu rẻ trước, LLM sau:
def classify(text):
t = text.lower()
if "hóa đơn" in t or "mã số thuế" in t and "cộng tiền hàng" in t:
return "hoa_don"
if "hợp đồng" in t or "điều khoản" in t or "bên a" in t:
return "hop_dong"
if "báo cáo" in t or "kỳ báo cáo" in t:
return "bao_cao"
return "khong_ro" # -> đẩy người phân loại
Từ khóa bắt được 80% ca rẻ và nhanh; phần khong_ro mới cần LLM hoặc con người. Đừng dùng LLM cho việc từ khóa làm được.
Orchestrator: một hàm điều phối rõ ràng
def run_pipeline(file_path):
text = ocr(preprocess_file(file_path)) # bài 2
doc_type = classify(text) # bài 7
if doc_type == "hoa_don":
data = extract_invoice(text) # bài 3
issues = validate_invoice(data, approved_msts) # bài 4
elif doc_type == "hop_dong":
data = extract_contract(text)
issues = check_clauses(text)
elif doc_type == "bao_cao":
data = extract_report(text)
issues = validate_report(data)
else:
return queue_for_review(file_path, ["KHONG_RO_LOAI"], "do")
lane = decide_lane(data.get("do_tin_cay", 0), issues)
if lane == "XANH":
return write_record(doc_type, data) # bài 5
return queue_for_review(data, issues, lane) # bài 6
Toàn bộ agent gói trong một hàm đọc-hiểu-được. Mỗi bước là một hàm đã xây ở bài trước.
Nguyên tắc: mỗi bước phải chống lỗi độc lập
Một bước hỏng không được làm sập cả pipeline. Bọc mỗi bước và ghi trạng thái:
try:
text = ocr(...)
except Exception as e:
return mark_failed(file_path, stage="ocr", error=str(e))
Khi lỗi, ta biết chính xác gãy ở bước nào — cực kỳ quan trọng khi vận hành thật.
Khung tư duy: Idempotent theo tài liệu, có thể chạy lại
Pipeline nên stateless theo file: cho cùng một file, chạy lại phải ra cùng kết quả và không gây tác dụng phụ trùng lặp (nhờ khóa idempotent ở bài 5). Nhờ đó bạn có thể chạy lại cả lô khi sửa lỗi mà không sợ hỏng dữ liệu.
Các bước lắp ráp
- Viết bước phân loại dùng từ khóa trước, LLM/người cho
khong_ro. - Xây orchestrator gọi tuần tự và rẽ nhánh theo loại.
- Bọc chống lỗi từng bước, ghi rõ stage khi thất bại.
- Đảm bảo idempotent theo file để chạy lại an toàn.
- Thêm logging + đo thời gian mỗi bước để tìm nút thắt.
- Chạy thử cả lô 50 file đầu-cuối, đo tỉ lệ Xanh/Vàng/Đỏ.
Checklist pipeline
- [ ] Có bước phân loại, có nhánh
khong_ro - [ ] Orchestrator rẽ nhánh theo loại tài liệu
- [ ] Mỗi bước bọc lỗi, ghi rõ stage gãy
- [ ] Idempotent theo file, chạy lại an toàn
- [ ] Log + đo thời gian từng bước
- [ ] Đã chạy thử cả lô, có số Xanh/Vàng/Đỏ
Sai lầm thường gặp
- Bỏ bước phân loại — hợp đồng lọt vào luồng hóa đơn, trích sai toàn bộ.
- Một lỗi làm sập cả lô — không bọc lỗi từng bước; một file hỏng dừng cả 500 file.
- Dùng LLM phân loại mọi file — tốn tiền và chậm cho việc từ khóa xử lý được.
- Không log stage — khi lỗi không biết gãy ở đâu, gỡ rối hàng giờ.