PricingLoginStart free
AI Agents · 课时

文档分类与路由

按类型对文档分类，并将其路由到专用代理处理器。

第 4 / 4 课13 个步骤

文档分类与路由 是 CoddyKit 上的免费 AI Agents 课时。 这是第 4 节课，共 4 节。 你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 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、服务协议和租赁合同。通过第二轮子类型分类，可以更精确地提取字段。

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 分类器进行微调或使用少样本提示。

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 与劳动合同的区分）可以更精确地提取字段。记录修正结果，通过反馈循环持续改进分类器。

免费开始

用 AI 导师学习 AI Agents — 免费

在浏览器中编写并运行真实代码，获得全天候 AI 导师的即时帮助，并在网页或应用中继续学习。

课程
60
课程
239
免费开始学习查看完整课程

常见问题解答

「文档分类与路由」课时是免费的吗？

是的 — 「文档分类与路由」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 AI Agents 课程的其余内容，请升级到 CoddyKit PRO。 AI Agents 课程共包含 4 节课。

「文档分类与路由」这节课中我会学到什么？

按类型对文档分类，并将其路由到专用代理处理器。 你通过在浏览器中直接运行的动手代码来练习 AI Agents，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

学习 AI Agents 需要有经验吗？

无需任何先前经验。CoddyKit 上的 AI Agents 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。 这是第 4 节课，共 4 节。

「文档分类与路由」课时需要多长时间？

大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。

我能在这节 AI Agents 课中编写并运行代码吗？

能。每节 AI Agents 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。

此课程中的所有课时

  1. 使用 PyMuPDF 和 pdfplumber 解析 PDF
  2. 扫描文档的 OCR
  3. 多文档问答代理
  4. 文档分类与路由
← 返回 AI Agents