在现代企业的订单处理流程中，不同类型的订单文档往往需要不同的处理逻辑和专业知识。传统的人工分类方式效率低下且容易出错，而基于规则的自动化系统又难以应对复杂多变的业务场景。本文将介绍如何利用大语言模型（LLM）构建一个高质量的订单文档分类器，实现智能路由到专业Agent的完整解决方案。

为什么需要文档分类器

在实际业务场景中，订单文档可能包含多种类型：

标准采购订单：常规商品采购，遵循标准流程
定制化订单：需要特殊定制的产品或服务
退货/换货订单：逆向物流处理
紧急订单：需要加急处理的高优先级订单
批量订单：大宗采购，涉及特殊定价和物流
国际订单：跨境交易，涉及海关和汇率

每种订单类型都需要不同的处理流程、审批权限和专业知识。一个高质量的分类器能够：

提升效率：自动路由减少人工干预
降低错误率：专业Agent处理对应领域问题
优化资源分配：将复杂任务分配给最合适的处理者
提升客户体验：更快的响应时间和更准确的处理

系统架构设计

一个完整的订单文档分类与路由系统通常包含以下核心组件：

┌─────────────────┐
│  订单文档输入    │
└────────┬────────┘
         │
         ▼
┌─────────────────┐
│  文档预处理      │  ← 提取关键信息、格式标准化
└────────┬────────┘
         │
         ▼
┌─────────────────┐
│  LLM分类器      │  ← 核心分类逻辑
└────────┬────────┘
         │
         ▼
┌─────────────────┐
│  置信度评估      │  ← 判断分类可靠性
└────────┬────────┘
         │
    ┌────┴────┐
    │         │
    ▼         ▼
 高置信度   低置信度
    │         │
    │         ▼
    │    ┌─────────────┐
    │    │ 人工复核队列 │
    │    └─────────────┘
    │
    ▼
┌─────────────────────────────────────┐
│        Agent路由器                   │
├─────────────────────────────────────┤
│  • 标准订单Agent                     │
│  • 定制订单Agent                     │
│  • 退货处理Agent                     │
│  • 紧急订单Agent                     │
│  • 批量订单Agent                     │
│  • 国际订单Agent                     │
└─────────────────────────────────────┘

核心技术实现

1. 文档预处理

在送入分类器之前，需要对订单文档进行标准化处理：

from typing import Dict, List, Any
import re
from datetime import datetime

class OrderDocumentPreprocessor:
    """订单文档预处理器"""

    def __init__(self):
        self.key_fields = [
            'order_id', 'customer_name', 'items',
            'total_amount', 'shipping_address', 'special_requirements'
        ]

    def extract_structured_data(self, raw_document: str) -> Dict[str, Any]:
        """
        从原始文档中提取结构化数据

        Args:
            raw_document: 原始订单文档文本

        Returns:
            结构化的订单数据字典
        """
        structured_data = {}

        # 提取订单号
        order_id_match = re.search(r'订单号[:：]\s*(\w+)', raw_document)
        if order_id_match:
            structured_data['order_id'] = order_id_match.group(1)

        # 提取金额信息
        amount_match = re.search(r'总金额[:：]\s*¥?([\d,]+\.?\d*)', raw_document)
        if amount_match:
            amount_str = amount_match.group(1).replace(',', '')
            structured_data['total_amount'] = float(amount_str)

        # 检测关键词
        structured_data['keywords'] = self._extract_keywords(raw_document)

        # 检测紧急程度
        structured_data['is_urgent'] = self._detect_urgency(raw_document)

        # 检测是否国际订单
        structured_data['is_international'] = self._detect_international(raw_document)

        return structured_data

    def _extract_keywords(self, text: str) -> List[str]:
        """提取关键特征词"""
        keywords = []

        # 定制化关键词
        if re.search(r'定制|客制化|个性化|特殊要求', text):
            keywords.append('customization')

        # 退货关键词
        if re.search(r'退货|退款|换货|退回', text):
            keywords.append('return')

        # 批量关键词
        if re.search(r'批量|大宗|批发|整批', text):
            keywords.append('bulk')

        return keywords

    def _detect_urgency(self, text: str) -> bool:
        """检测是否紧急订单"""
        urgent_patterns = [
            r'紧急', r'加急', r'特急', r'URGENT', r'ASAP',
            r'立即', r'马上', r'尽快'
        ]
        return any(re.search(pattern, text, re.IGNORECASE) for pattern in urgent_patterns)

    def _detect_international(self, text: str) -> bool:
        """检测是否国际订单"""
        international_indicators = [
            r'国际', r'海外', r'出口', r'进口', r'跨境',
            r'关税', r'清关', r'international', r'overseas'
        ]
        return any(re.search(pattern, text, re.IGNORECASE) for pattern in international_indicators)

# generated by AI

2. LLM分类器核心实现

使用LLM进行分类的关键在于设计高质量的提示词（Prompt）：

from typing import Tuple
import anthropic
import json

class OrderClassifier:
    """基于LLM的订单文档分类器"""

    # 定义订单类型
    ORDER_TYPES = {
        'standard': '标准采购订单',
        'customized': '定制化订单',
        'return': '退货/换货订单',
        'urgent': '紧急订单',
        'bulk': '批量订单',
        'international': '国际订单'
    }

    def __init__(self, api_key: str, model: str = "claude-3-5-sonnet-20241022"):
        self.client = anthropic.Anthropic(api_key=api_key)
        self.model = model

    def classify(self, document: str, structured_data: Dict[str, Any]) -> Tuple[str, float, str]:
        """
        对订单文档进行分类

        Args:
            document: 原始订单文档
            structured_data: 预处理后的结构化数据

        Returns:
            (order_type, confidence, reasoning): 订单类型、置信度、推理过程
        """
        prompt = self._build_classification_prompt(document, structured_data)

        message = self.client.messages.create(
            model=self.model,
            max_tokens=1024,
            messages=[{
                "role": "user",
                "content": prompt
            }]
        )

        # 解析响应
        response_text = message.content[0].text
        result = self._parse_classification_result(response_text)

        return result

    def _build_classification_prompt(self, document: str, structured_data: Dict[str, Any]) -> str:
        """构建分类提示词"""

        prompt = f"""你是一个专业的订单文档分类专家。请仔细分析以下订单文档，并将其分类到最合适的类别中。

## 订单类型定义

1. **标准采购订单** (standard)
   - 常规商品采购
   - 遵循标准流程
   - 无特殊要求

2. **定制化订单** (customized)
   - 需要产品定制
   - 有特殊规格要求
   - 非标准产品

3. **退货/换货订单** (return)
   - 商品退回
   - 换货请求
   - 售后处理

4. **紧急订单** (urgent)
   - 需要加急处理
   - 有明确的紧急时间要求
   - 高优先级

5. **批量订单** (bulk)
   - 大批量采购
   - 涉及特殊定价
   - 可能需要特殊物流

6. **国际订单** (international)
   - 跨境交易
   - 涉及海关清关
   - 需要处理汇率和国际物流

## 订单文档

{document}

## 预提取的结构化信息

```json
{json.dumps(structured_data, ensure_ascii=False, indent=2)}

任务要求

请分析上述订单文档，并按照以下JSON格式返回分类结果：

{{
  "order_type": "订单类型代码 (standard/customized/return/urgent/bulk/international)",
  "confidence": 0.0-1.0之间的置信度分数,
  "reasoning": "详细的分类推理过程，说明为什么选择这个类别",
  "key_factors": ["关键判断因素1", "关键判断因素2", ...]
}}

注意：

如果订单同时符合多个类型，选择最主要的特征类型
紧急订单和国际订单具有更高的优先级
置信度应该反映你对分类结果的确定程度
推理过程要具体，引用文档中的关键信息 """ return prompt
def _parse_classification_result(self, response: str) -> Tuple[str, float, str]: “““解析LLM返回的分类结果””” try: # 提取JSON部分 json_match = re.search(r’{[\s\S]*}’, response) if json_match: result = json.loads(json_match.group()) return ( result[‘order_type’], float(result[‘confidence’]), result[‘reasoning’] ) else: # 如果无法解析，返回默认值 return (‘standard’, 0.5, ‘无法解析分类结果’) except Exception as e: print(f"解析错误: {e}") return (‘standard’, 0.3, f’解析失败: {str(e)}’)

generated by AI


### 3. 置信度评估与人工复核

并非所有分类结果都需要相同的处理方式。设置合理的置信度阈值可以平衡自动化程度和准确率：

```python
class ConfidenceEvaluator:
    """置信度评估器"""

    def __init__(self, auto_route_threshold: float = 0.85, manual_review_threshold: float = 0.60):
        self.auto_route_threshold = auto_route_threshold
        self.manual_review_threshold = manual_review_threshold

    def evaluate(self, order_type: str, confidence: float, structured_data: Dict[str, Any]) -> str:
        """
        评估分类结果并决定处理策略

        Args:
            order_type: 分类的订单类型
            confidence: LLM给出的置信度
            structured_data: 订单结构化数据

        Returns:
            处理策略: 'auto_route', 'manual_review', 或 'reject'
        """
        # 高置信度：直接自动路由
        if confidence >= self.auto_route_threshold:
            return 'auto_route'

        # 中等置信度：考虑订单金额等因素
        elif confidence >= self.manual_review_threshold:
            # 高价值订单即使中等置信度也需要人工复核
            if structured_data.get('total_amount', 0) > 100000:
                return 'manual_review'

            # 国际订单涉及复杂流程，建议人工复核
            if structured_data.get('is_international', False):
                return 'manual_review'

            # 其他中等置信度情况可以自动路由
            return 'auto_route'

        # 低置信度：必须人工复核
        else:
            return 'manual_review'

    def get_review_priority(self, order_type: str, confidence: float, structured_data: Dict[str, Any]) -> int:
        """
        计算人工复核的优先级（1-5，5最高）
        """
        priority = 3  # 默认中等优先级

        # 紧急订单提高优先级
        if structured_data.get('is_urgent', False):
            priority += 2

        # 低置信度提高优先级
        if confidence < 0.5:
            priority += 1

        # 高金额提高优先级
        if structured_data.get('total_amount', 0) > 50000:
            priority += 1

        return min(priority, 5)  # 最高5级

# generated by AI

4. Agent路由器实现

分类完成后，需要将订单路由到对应的专业Agent：

from abc import ABC, abstractmethod
from typing import Protocol

class OrderAgent(Protocol):
    """订单处理Agent接口"""

    def can_handle(self, order_type: str) -> bool:
        """判断是否能处理该类型订单"""
        ...

    def process(self, document: str, structured_data: Dict[str, Any]) -> Dict[str, Any]:
        """处理订单"""
        ...

class AgentRouter:
    """Agent路由器"""

    def __init__(self):
        self.agents: Dict[str, OrderAgent] = {}
        self.fallback_agent: OrderAgent = None

    def register_agent(self, order_type: str, agent: OrderAgent):
        """注册专业Agent"""
        self.agents[order_type] = agent
        print(f"已注册 {order_type} 类型的Agent")

    def set_fallback_agent(self, agent: OrderAgent):
        """设置默认/兜底Agent"""
        self.fallback_agent = agent

    def route(self, order_type: str, document: str, structured_data: Dict[str, Any]) -> Dict[str, Any]:
        """
        路由订单到对应的Agent

        Args:
            order_type: 订单类型
            document: 原始文档
            structured_data: 结构化数据

        Returns:
            处理结果
        """
        agent = self.agents.get(order_type, self.fallback_agent)

        if agent is None:
            raise ValueError(f"没有找到处理 {order_type} 类型订单的Agent，且未设置fallback agent")

        print(f"路由订单到 {order_type} Agent...")
        result = agent.process(document, structured_data)

        return {
            'order_type': order_type,
            'agent': agent.__class__.__name__,
            'processing_result': result,
            'timestamp': datetime.now().isoformat()
        }

# 示例：标准订单Agent
class StandardOrderAgent:
    """标准订单处理Agent"""

    def can_handle(self, order_type: str) -> bool:
        return order_type == 'standard'

    def process(self, document: str, structured_data: Dict[str, Any]) -> Dict[str, Any]:
        """处理标准订单"""
        print("处理标准订单...")

        # 验证库存
        inventory_check = self._check_inventory(structured_data)

        # 计算价格
        pricing = self._calculate_pricing(structured_data)

        # 安排物流
        shipping = self._arrange_shipping(structured_data)

        return {
            'status': 'processed',
            'inventory_check': inventory_check,
            'pricing': pricing,
            'shipping': shipping
        }

    def _check_inventory(self, data: Dict[str, Any]) -> Dict[str, Any]:
        """检查库存（示例）"""
        return {'available': True, 'estimated_ship_date': '2025-12-22'}

    def _calculate_pricing(self, data: Dict[str, Any]) -> Dict[str, Any]:
        """计算价格（示例）"""
        amount = data.get('total_amount', 0)
        return {'total': amount, 'discount': 0, 'final_amount': amount}

    def _arrange_shipping(self, data: Dict[str, Any]) -> Dict[str, Any]:
        """安排物流（示例）"""
        return {'carrier': '顺丰速运', 'estimated_delivery': '2-3工作日'}

# 示例：定制订单Agent
class CustomizedOrderAgent:
    """定制订单处理Agent"""

    def can_handle(self, order_type: str) -> bool:
        return order_type == 'customized'

    def process(self, document: str, structured_data: Dict[str, Any]) -> Dict[str, Any]:
        """处理定制订单"""
        print("处理定制订单...")

        # 评估定制需求
        requirements = self._evaluate_requirements(document)

        # 生成报价
        quotation = self._generate_quotation(requirements, structured_data)

        # 评估生产周期
        production_time = self._estimate_production_time(requirements)

        return {
            'status': 'requires_confirmation',
            'requirements': requirements,
            'quotation': quotation,
            'production_time': production_time
        }

    def _evaluate_requirements(self, document: str) -> Dict[str, Any]:
        """评估定制需求（示例）"""
        return {'complexity': 'medium', 'feasibility': 'high'}

    def _generate_quotation(self, requirements: Dict, data: Dict) -> Dict[str, Any]:
        """生成报价（示例）"""
        base_amount = data.get('total_amount', 0)
        customization_fee = base_amount * 0.3
        return {'base': base_amount, 'customization_fee': customization_fee, 'total': base_amount + customization_fee}

    def _estimate_production_time(self, requirements: Dict) -> str:
        """评估生产周期（示例）"""
        return '15-20工作日'

# generated by AI

5. 完整流程编排

将所有组件整合到一起：

class OrderProcessingPipeline:
    """订单处理流水线"""

    def __init__(self, api_key: str):
        self.preprocessor = OrderDocumentPreprocessor()
        self.classifier = OrderClassifier(api_key)
        self.confidence_evaluator = ConfidenceEvaluator()
        self.router = AgentRouter()

        # 注册各类Agent
        self._setup_agents()

    def _setup_agents(self):
        """初始化并注册所有Agent"""
        self.router.register_agent('standard', StandardOrderAgent())
        self.router.register_agent('customized', CustomizedOrderAgent())
        # 注册其他Agent...

        # 设置fallback agent
        self.router.set_fallback_agent(StandardOrderAgent())

    def process(self, raw_document: str) -> Dict[str, Any]:
        """
        完整的订单处理流程

        Args:
            raw_document: 原始订单文档

        Returns:
            处理结果
        """
        # 步骤1: 预处理
        print("步骤1: 预处理订单文档...")
        structured_data = self.preprocessor.extract_structured_data(raw_document)

        # 步骤2: 分类
        print("步骤2: 分类订单类型...")
        order_type, confidence, reasoning = self.classifier.classify(raw_document, structured_data)
        print(f"  分类结果: {order_type}")
        print(f"  置信度: {confidence:.2f}")
        print(f"  推理: {reasoning}")

        # 步骤3: 置信度评估
        print("步骤3: 评估置信度...")
        strategy = self.confidence_evaluator.evaluate(order_type, confidence, structured_data)
        print(f"  处理策略: {strategy}")

        # 步骤4: 根据策略处理
        if strategy == 'manual_review':
            priority = self.confidence_evaluator.get_review_priority(order_type, confidence, structured_data)
            return {
                'status': 'pending_review',
                'order_type': order_type,
                'confidence': confidence,
                'review_priority': priority,
                'reasoning': reasoning
            }

        # 步骤5: 自动路由到Agent
        print("步骤4: 路由到专业Agent...")
        result = self.router.route(order_type, raw_document, structured_data)

        return {
            'status': 'auto_processed',
            'classification': {
                'order_type': order_type,
                'confidence': confidence,
                'reasoning': reasoning
            },
            'processing': result
        }

# 使用示例
if __name__ == "__main__":
    # 示例订单文档
    sample_order = """
    订单号: ORD-2025-12345
    客户名称: 张三科技有限公司
    订单日期: 2025-12-20

    订单内容:
    - 产品A x 100件
    - 产品B x 50件

    总金额: ¥58,000

    特殊要求: 需要在产品A上印制公司Logo

    配送地址: 上海市浦东新区XX路XX号
    联系电话: 138-1234-5678
    """

    # 初始化处理流水线
    pipeline = OrderProcessingPipeline(api_key="your-api-key")

    # 处理订单
    result = pipeline.process(sample_order)

    print("\n===== 最终结果 =====")
    print(json.dumps(result, ensure_ascii=False, indent=2))

# generated by AI

最佳实践与优化建议

1. 提示词工程

明确类型定义：为每个订单类型提供清晰、具体的定义和示例
提供上下文：包含业务背景和领域知识
要求结构化输出：使用JSON格式便于解析
要求推理过程：让模型解释分类依据，提高可解释性

2. Few-shot Learning

对于特定领域或复杂场景，可以在提示词中加入示例：

def _build_classification_prompt_with_examples(self, document: str) -> str:
    examples = """
## 分类示例

### 示例1: 定制化订单
文档: "订单号：001，需要在T恤上印制特定图案，尺寸为XXL，数量100件"
分类: customized
原因: 明确提到"印制特定图案"，属于产品定制需求

### 示例2: 紧急订单
文档: "订单号：002，标准商品100件，总金额5000元，备注：需要明天发货，加急！"
分类: urgent
原因: 明确要求"明天发货"和"加急"标识

### 示例3: 批量订单
文档: "订单号：003，批发采购标准商品5000件，总金额50万元，需要特殊报价"
分类: bulk
原因: 数量大（5000件）、金额高（50万），且提到"批发"和"特殊报价"
"""
    # 将examples插入到prompt中
    return examples + "\n\n" + "现在请分类以下订单：\n" + document

# generated by AI

3. 多模型集成

对于关键业务，可以使用多个模型进行ensemble：

class EnsembleClassifier:
    """集成分类器"""

    def __init__(self, classifiers: List[OrderClassifier]):
        self.classifiers = classifiers

    def classify(self, document: str, structured_data: Dict[str, Any]) -> Tuple[str, float, str]:
        """使用多个分类器进行投票"""
        results = []

        for classifier in self.classifiers:
            order_type, confidence, reasoning = classifier.classify(document, structured_data)
            results.append((order_type, confidence, reasoning))

        # 投票决策
        type_votes = {}
        for order_type, confidence, _ in results:
            if order_type not in type_votes:
                type_votes[order_type] = []
            type_votes[order_type].append(confidence)

        # 选择得票最多且平均置信度最高的类型
        best_type = max(type_votes.items(), key=lambda x: (len(x[1]), sum(x[1])/len(x[1])))[0]
        avg_confidence = sum(type_votes[best_type]) / len(type_votes[best_type])

        return best_type, avg_confidence, f"集成分类结果，基于{len(self.classifiers)}个模型"

# generated by AI

4. 持续学习与优化

建立反馈循环，不断优化分类器性能：

class ClassificationFeedbackSystem:
    """分类反馈系统"""

    def __init__(self, db_connection):
        self.db = db_connection

    def record_classification(self, document: str, predicted_type: str,
                             confidence: float, actual_type: str = None):
        """记录分类结果和实际标签"""
        self.db.insert({
            'document': document,
            'predicted_type': predicted_type,
            'confidence': confidence,
            'actual_type': actual_type,
            'timestamp': datetime.now(),
            'correct': predicted_type == actual_type if actual_type else None
        })

    def get_accuracy_metrics(self) -> Dict[str, float]:
        """计算准确率指标"""
        records = self.db.query("SELECT * FROM classifications WHERE actual_type IS NOT NULL")

        total = len(records)
        correct = sum(1 for r in records if r['correct'])

        # 按类型统计
        type_metrics = {}
        for order_type in OrderClassifier.ORDER_TYPES.keys():
            type_records = [r for r in records if r['predicted_type'] == order_type]
            if type_records:
                type_correct = sum(1 for r in type_records if r['correct'])
                type_metrics[order_type] = {
                    'accuracy': type_correct / len(type_records),
                    'count': len(type_records)
                }

        return {
            'overall_accuracy': correct / total if total > 0 else 0,
            'total_classifications': total,
            'type_metrics': type_metrics
        }

    def identify_improvement_areas(self) -> List[str]:
        """识别需要改进的领域"""
        metrics = self.get_accuracy_metrics()
        improvement_areas = []

        for order_type, stats in metrics['type_metrics'].items():
            if stats['accuracy'] < 0.85:  # 准确率阈值
                improvement_areas.append(
                    f"{order_type}: 准确率 {stats['accuracy']:.2%}，需要优化"
                )

        return improvement_areas

# generated by AI

5. 性能优化

对于高吞吐量场景，考虑以下优化：

批处理：将多个订单合并为一次API调用
缓存：对于相似文档，缓存分类结果
异步处理：使用消息队列解耦分类和路由
模型选择：根据延迟要求选择合适规模的模型

import asyncio
from typing import List

class AsyncOrderClassifier:
    """异步订单分类器"""

    async def classify_batch(self, documents: List[str]) -> List[Tuple[str, float, str]]:
        """批量异步分类"""
        tasks = [self.classify_async(doc) for doc in documents]
        results = await asyncio.gather(*tasks)
        return results

    async def classify_async(self, document: str) -> Tuple[str, float, str]:
        """单个文档异步分类"""
        # 实现异步分类逻辑
        pass

# generated by AI

监控与可观测性

生产环境中，需要建立完善的监控体系：

from dataclasses import dataclass
from typing import Optional
import time

@dataclass
class ClassificationMetrics:
    """分类指标"""
    document_id: str
    order_type: str
    confidence: float
    processing_time: float
    strategy: str
    timestamp: float

class MetricsCollector:
    """指标收集器"""

    def __init__(self):
        self.metrics: List[ClassificationMetrics] = []

    def record_classification(self, metric: ClassificationMetrics):
        """记录分类指标"""
        self.metrics.append(metric)

        # 实时监控
        self._check_alerts(metric)

    def _check_alerts(self, metric: ClassificationMetrics):
        """检查告警条件"""
        # 低置信度告警
        if metric.confidence < 0.6:
            self._send_alert(f"低置信度分类: {metric.order_type}, 置信度: {metric.confidence}")

        # 处理时间告警
        if metric.processing_time > 5.0:  # 5秒
            self._send_alert(f"分类耗时过长: {metric.processing_time}秒")

    def _send_alert(self, message: str):
        """发送告警"""
        print(f"[ALERT] {message}")
        # 实际应该集成到告警系统

    def get_summary(self, time_window: int = 3600) -> Dict[str, Any]:
        """获取指标摘要"""
        now = time.time()
        recent_metrics = [m for m in self.metrics if now - m.timestamp < time_window]

        if not recent_metrics:
            return {}

        return {
            'total_classifications': len(recent_metrics),
            'avg_confidence': sum(m.confidence for m in recent_metrics) / len(recent_metrics),
            'avg_processing_time': sum(m.processing_time for m in recent_metrics) / len(recent_metrics),
            'auto_route_rate': sum(1 for m in recent_metrics if m.strategy == 'auto_route') / len(recent_metrics),
            'type_distribution': self._get_type_distribution(recent_metrics)
        }

    def _get_type_distribution(self, metrics: List[ClassificationMetrics]) -> Dict[str, int]:
        """获取类型分布"""
        distribution = {}
        for metric in metrics:
            distribution[metric.order_type] = distribution.get(metric.order_type, 0) + 1
        return distribution

# generated by AI

总结

构建高质量的订单文档分类器需要综合考虑多个方面：

准确性：通过精心设计的提示词、预处理和后处理提升分类准确率
可靠性：设置合理的置信度阈值，确保关键订单有人工把关
效率：使用异步处理、批处理等技术提升吞吐量
可维护性：建立反馈机制，持续优化系统性能
可观测性：完善的监控和告警系统，及时发现问题

通过LLM驱动的智能分类器，配合专业的Agent路由系统，可以显著提升订单处理的自动化水平和准确性。随着业务的发展，系统可以不断学习和优化，适应新的订单类型和业务场景。

构建高质量订单文档分类器：智能导流到专业Agent

从意图识别到精准路由的完整解决方案

目录: