文書の分類とルーティング
種類ごとに文書を分類し、専門のエージェントハンドラーへ振り分けます。
「文書の分類とルーティング」はCoddyKit上の無料AI Agentsレッスンです。 これはレッスン4/4です。 下記で完全なレッスンを無料で読むことができます。その後、ブラウザ内の組み込みコードエディタと24時間対応のAIチューターでハンズオン演習できます。 これはAI Agents学習パスの一部であり、ウェブとCoddyKitアプリ全体で進捗が同期されます。 AI Agentsコースには全4レッスンが含まれています。
文書分類が重要な理由
ドキュメントインテリジェンスエージェントには、請求書、契約書、レポート、メール、領収書など、さまざまな種類の文書が入力されます。それぞれの種類には、異なる抽出ロジックとビジネスルールが必要です。
文書分類は、以降の処理を行う前に各文書を適切なハンドラーへ振り分けます。これはエージェントの受付ロジックとして機能します。
LLMベースの分類
最も単純で柔軟な分類器はLLMを使用します。文書テキストの一部を渡し、文書の種類を特定するようモデルに依頼します。文書の種類が明確に異なる場合に適しています。
import openai
import os
client = openai.OpenAI(api_key=os.getenv('OPENAI_API_KEY'))
DOC_TYPES = ['invoice', 'contract', 'report', 'email', 'receipt', 'form', 'letter', 'other']
CLASSIFY_PROMPT = '''Classify this document into one of these types: {types}
Document excerpt (first 1000 characters):
{text}
Respond with ONLY the document type as a single word from the list above.'''
def classify_with_llm(text):
response = client.chat.completions.create(
model='gpt-4o-mini', # cheap and fast for classification
messages=[{
'role': 'user',
'content': CLASSIFY_PROMPT.format(
types=', '.join(DOC_TYPES),
text=text[:1000]
)
}],
max_tokens=10,
temperature=0
)
predicted = response.choices[0].message.content.strip().lower()
return predicted if predicted in DOC_TYPES else 'other'ルールベース分類へのフォールバック
LLMによる分類は高精度ですが、費用がかかり、レイテンシーも増加します。一般的で明確に定義された文書の種類であれば、キーワードによるルールベース分類器は高速、無料、かつ解釈可能です。
ルールベース分類を高速パスとして使用し、判断が難しい場合はLLMにフォールバックします。
CLASSIFICATION_RULES = {
'invoice': [
'invoice', 'invoice number', 'bill to', 'amount due',
'total amount', 'tax invoice', 'payment terms'
],
'contract': [
'agreement', 'terms and conditions', 'hereby agrees',
'party a', 'party b', 'whereas', 'obligations'
],
'report': [
'executive summary', 'quarterly report', 'annual report',
'findings', 'recommendations', 'methodology'
],
'email': ['from:', 'to:', 'subject:', 'date:', 'dear ', 'regards,'],
'receipt': ['receipt', 'thank you for your purchase', 'transaction id', 'cashier']
}
def classify_with_rules(text):
text_lower = text.lower()
scores = {}
for doc_type, keywords in CLASSIFICATION_RULES.items():
score = sum(1 for kw in keywords if kw in text_lower)
if score > 0:
scores[doc_type] = score
if not scores:
return None # no match — fall through to LLM
return max(scores, key=scores.get)
if __name__ == '__main__':
demo_text = 'INVOICE\nBill To: Acme Corp\nAmount Due: $500\nPayment Terms: Net 30'
print('Classified as:', classify_with_rules(demo_text))
段階的な分類戦略
ルールベース分類とLLM分類を段階的なアプローチで組み合わせます。最初に高速なルールを適用し、ルールで判断できない場合にのみLLMを使用します。これにより、エッジケースの精度を維持しながらコストを最小限に抑えられます。
def classify_document(text):
# Tier 1: rule-based (free, fast)
result = classify_with_rules(text)
if result:
print(f'Rule-based classification: {result}')
return {'type': result, 'method': 'rules', 'confidence': None}
# Tier 2: LLM (accurate, slower)
result = classify_with_llm(text)
print(f'LLM classification: {result}')
return {'type': result, 'method': 'llm', 'confidence': None}信頼度の閾値
すべての分類結果の信頼度が同じとは限りません。LLM分類器には、分類結果とともに信頼度スコアを返すよう依頼します。信頼度が低い場合は、人による確認のために文書へフラグを付けます。
import json
CLASSIFY_WITH_CONFIDENCE_PROMPT = '''Classify this document. Return JSON:
{{"type": "invoice", "confidence": 0.95, "reason": "Contains invoice number and payment terms"}}
Valid types: {types}
Document excerpt: {text}
JSON:'''
def classify_with_confidence(text):
response = client.chat.completions.create(
model='gpt-4o-mini',
messages=[{'role': 'user', 'content': CLASSIFY_WITH_CONFIDENCE_PROMPT.format(
types=', '.join(DOC_TYPES),
text=text[:1000]
)}],
temperature=0
)
try:
result = json.loads(response.choices[0].message.content)
return result
except json.JSONDecodeError:
return {'type': 'other', 'confidence': 0.0, 'reason': 'Parse error'}
LOW_CONFIDENCE_THRESHOLD = 0.6
def classify_and_check(text):
result = classify_with_confidence(text)
if result['confidence'] < LOW_CONFIDENCE_THRESHOLD:
result['needs_review'] = True
print(f'Low confidence ({result["confidence"]}) — flagging for review')
return result文書ルーター
分類が完了すると、ルーターが文書を専用のハンドラーへ振り分けます。各ハンドラーは、その文書の種類に関連する特定のフィールドを抽出する方法を把握しています。
def handle_invoice(text):
# Extract: vendor, invoice number, total, due date
extract_prompt = f'''Extract from this invoice (JSON):
{{"vendor": "", "invoice_number": "", "total": 0, "due_date": "", "line_items": []}}
{text[:2000]}\n\nJSON:'''
return llm_call(extract_prompt)
def handle_contract(text):
# Extract: parties, effective date, term, key obligations
extract_prompt = f'''Extract from this contract (JSON):
{{"parties": [], "effective_date": "", "term_months": 0, "key_obligations": []}}
{text[:2000]}\n\nJSON:'''
return llm_call(extract_prompt)
ROUTERS = {
'invoice': handle_invoice,
'contract': handle_contract,
'report': lambda t: llm_call(f'Summarize this report in 3 bullet points:\n{t[:2000]}'),
'email': lambda t: llm_call(f'Extract: sender, subject, action required from this email:\n{t[:1000]}')
}
def route_document(text):
classification = classify_document(text)
doc_type = classification['type']
handler = ROUTERS.get(doc_type, lambda t: llm_call(f'Describe this document:\n{t[:1000]}'))
return handler(text)サブタイプ分類
「契約書」のような上位レベルの種類には、雇用契約、NDA、サービス契約、賃貸借契約などのサブタイプがあります。2回目のサブタイプ分類を行うことで、より正確なフィールド抽出が可能になります。
CONTRACT_SUBTYPES = {
'employment': ['employment', 'employee', 'employer', 'salary', 'compensation', 'job title'],
'nda': ['non-disclosure', 'confidential', 'nda', 'proprietary information'],
'service': ['service agreement', 'scope of work', 'deliverables', 'milestone'],
'lease': ['lease', 'landlord', 'tenant', 'rent', 'premises', 'square feet']
}
def classify_contract_subtype(text):
text_lower = text.lower()
scores = {
subtype: sum(1 for kw in keywords if kw in text_lower)
for subtype, keywords in CONTRACT_SUBTYPES.items()
}
best = max(scores, key=scores.get)
if scores[best] == 0:
return 'general'
return best
def handle_contract_routed(text):
subtype = classify_contract_subtype(text)
print(f'Contract subtype: {subtype}')
# Route to specialized extractor
if subtype == 'nda':
return extract_nda_fields(text)
elif subtype == 'employment':
return extract_employment_fields(text)
else:
return handle_contract(text)バッチ分類パイプライン
本番環境では、文書はバッチで到着します。まずすべての文書を分類し、種類ごとにグループ化してから、各グループを並列に処理すると効率的です。
from concurrent.futures import ThreadPoolExecutor
import time
def classify_batch(documents):
results = []
for doc in documents:
text = extract_text(doc['path']) # PDF, OCR, or plain text
classification = classify_document(text[:1500])
results.append({
'doc_id': doc['id'],
'path': doc['path'],
'type': classification['type'],
'method': classification['method'],
'text': text
})
return results
def process_batch(documents, max_workers=4):
# Step 1: classify all (fast)
classified = classify_batch(documents)
# Step 2: group by type
from collections import defaultdict
by_type = defaultdict(list)
for doc in classified:
by_type[doc['type']].append(doc)
# Step 3: process each group
all_results = {}
for doc_type, docs in by_type.items():
handler = ROUTERS.get(doc_type)
if handler:
with ThreadPoolExecutor(max_workers=max_workers) as executor:
futures = {executor.submit(handler, d['text']): d for d in docs}
for fut, doc in futures.items():
all_results[doc['doc_id']] = fut.result()
return all_results「その他」および不明な種類への対応
「その他」に分類された文書や信頼度の低い文書には、フォールバック戦略が必要です。人による確認のためにフラグを付ける、汎用的な抽出を試みる、文書の種類をユーザーに尋ねる、といった方法があります。
HUMAN_REVIEW_QUEUE = []
def process_document(doc_path):
text = extract_text(doc_path)
classification = classify_with_confidence(text[:1500])
# High confidence path
if classification['confidence'] >= 0.8 and classification['type'] != 'other':
handler = ROUTERS.get(classification['type'])
return {
'result': handler(text),
'type': classification['type'],
'auto_processed': True
}
# Low confidence or unknown type
HUMAN_REVIEW_QUEUE.append({
'path': doc_path,
'predicted_type': classification['type'],
'confidence': classification['confidence'],
'reason': classification.get('reason', '')
})
print(f'Added to review queue: {doc_path} ({classification["confidence"]:.0%} confident)')
return {'auto_processed': False, 'queued_for_review': True}分類のフィードバックループ
人が誤分類を修正したら、その修正内容を記録します。これらのログを使ってルールベース分類器を改善し、時間の経過とともにLLM分類器をファインチューニングしたり、few-shotプロンプトを使ったりします。
correction_log = []
def log_correction(doc_path, predicted_type, correct_type, text_sample):
correction_log.append({
'doc_path': doc_path,
'predicted': predicted_type,
'correct': correct_type,
'text_sample': text_sample[:200]
})
print(f'Logged correction: {predicted_type} -> {correct_type}')
def build_few_shot_examples(n=5):
recent = correction_log[-n:] # use most recent corrections
examples = []
for entry in recent:
examples.append(
f'Text: {entry["text_sample"]}\nCorrect type: {entry["correct"]}'
)
return '\n\n'.join(examples)
def classify_with_few_shot(text):
few_shot = build_few_shot_examples()
prompt = f'''Examples of correct classifications:\n{few_shot}\n\nNow classify:\n{text[:800]}\n\nType:'''
return llm_call(prompt).strip().lower()分類後の構造化データ抽出
文書の種類が決まったら、その種類で重要な特定のフィールドを抽出します。JSONスキーマを使ったLLMの構造化抽出を利用すると、一貫性のある解析可能な出力を得られます。
import json
class FakeMsg:
def __init__(self, content): self.content = content
class FakeChoice:
def __init__(self, content): self.message = FakeMsg(content)
class FakeResponse:
def __init__(self, content): self.choices = [FakeChoice(content)]
class _Completions:
@staticmethod
def create(model, messages, temperature):
return FakeResponse('{"vendor_name": "Acme", "invoice_number": "123", "total": 100}')
class _Chat:
completions = _Completions()
class FakeClient:
chat = _Chat()
client = FakeClient()
EXTRACTION_SCHEMAS = {
'invoice': '{vendor_name: , invoice_number: , invoice_date: , due_date: , subtotal: 0, tax: 0, total: 0, line_items: []}',
}
def extract_structured_fields(text, doc_type):
schema = EXTRACTION_SCHEMAS.get(doc_type)
if not schema:
return {'error': f'No extraction schema for type: {doc_type}'}
prompt = (f'Extract fields from this {doc_type}. Return JSON matching this schema:\n'
f'{schema}\n\nDocument:\n{text[:2000]}\n\nJSON:')
response = client.chat.completions.create(model='gpt-4o-mini', messages=[{'role': 'user', 'content': prompt}], temperature=0)
try:
return json.loads(response.choices[0].message.content)
except json.JSONDecodeError:
return {'error': 'Could not parse extraction result'}
print(extract_structured_fields('Invoice #123 from Acme for $100', 'invoice'))理解度チェック
段階的な分類戦略（最初にルールを適用し、LLMをフォールバックとして使用する）の利点は何ですか。
復習：文書分類とルーティング
文書分類は、入力された文書を専用のハンドラーへ振り分けます。段階的なアプローチを使用し、一般的な種類には高速なキーワードルールを、エッジケースには信頼度スコア付きのLLM分類を、信頼度の低い文書には人による確認キューを適用します。
各文書の種類には専用の抽出ハンドラーを用意します。サブタイプ分類（例：NDAと雇用契約の区別）により、より正確なフィールド抽出が可能になります。修正内容を記録し、フィードバックループを通じて時間の経過とともに分類器を改善します。
よくある質問
「文書の分類とルーティング」レッスンは無料ですか？
はい。「文書の分類とルーティング」の完全なテキストはこのウェブで無料で読めます。インタラクティブに演習し（組み込みコードエディタと24時間対応のAIチューター）、AI Agentsコースの残りをアンロックするには、CoddyKit PROにアップグレードしてください。 AI Agentsコースには全4レッスンが含まれています。
「文書の分類とルーティング」で何を学びますか？
種類ごとに文書を分類し、専門のエージェントハンドラーへ振り分けます。 ブラウザで直接実行するハンズオンコードでAI Agentsを演習し、24時間対応のAIチューターがレッスンを進める中での質問に答えます。
AI Agentsを始めるのに経験は必要ですか？
事前経験は必要ありません。CoddyKitのAI Agentsは初級者から上級者向けに構成されているため、ここから始めるか最初から始めて、自分のペースで進むことができます。 これはレッスン4/4です。
「文書の分類とルーティング」レッスンにはどのくらい時間がかかりますか？
ほとんどのCoddyKitレッスンは約5～10分かかります。各レッスンはコンパクトでインタラクティブなので、着実に進歩し、ウェブとアプリ全体で正確に前回の場所から再開できます。
このAI Agentsレッスンでコードを書いて実行できますか？
はい。すべてのAI Agentsレッスンに組み込みコードエディタが含まれているため、ブラウザでリアルコードを書いて実行し、即座のAIフィードバックを取得できます。ローカル設定は不要です。