基于混元大模型+RPA处理非结构化业务文档：自动化流程搭建与落地

┌─────────────────────────────────────────┐
│           感知层：文档输入                 │
│  扫描仪/PDF目录 → 混元大模型OCR+理解      │
└──────────────┬──────────────────────────┘
               │ 结构化JSON
┌──────────────▼──────────────────────────┐
│           决策层：逻辑判断                 │
│  合同类型识别 → 字段提取 → 路由决策        │
└──────────────┬──────────────────────────┘
               │ API触发
┌──────────────▼──────────────────────────┐
│           执行层：RPA落地                  │
│  登录财务系统 → 录入数据 → 发送通知        │
└─────────────────────────────────────────┘

import base64
import requests
from pydantic import BaseModel, ValidationError

# 注意：以下代码依赖 Pydantic v2（pip install pydantic>=2.0）
class ContractSchema(BaseModel):
    party_a: str
    party_b: str
    amount: float
    pay_method: str
    milestones: list

def encode_image(image_path: str) -> str:
    """将图片转为 Base64 编码，用于混元 API 上传"""
    with open(image_path, "rb") as f:
        return base64.b64encode(f.read()).decode("utf-8")

def parse_contract(image_path: str) -> dict:
    try:
        resp = requests.post(
            "https://hunyuan.tencentcloudapi.com/v1/chat/completions",
            headers={"Authorization": "Bearer YOUR_API_KEY"},
            json={
                "model": "hunyuan-vision",
                "messages": [{
                    "role": "user",
                    "content": [
                        {
                            "type": "text", 
                            "text": (
                                "你是一位合同审核助手。请从以下合同文本中提取关键信息，"
                                "输出严格JSON格式：{\"甲方\":\"\",\"乙方\":\"\",\"合同金额\":0,"
                                "\"付款方式\":\"\",\"关键节点\":[]}。金额转数字，"
                                "付款方式标准化为月付/季付/年付/一次性。"
                            )
                        },
                        {
                            "type": "image_url", 
                            "image_url": {
                                "url": f"data:image/png;base64,{encode_image(image_path)}"
                            }
                        }
                    ]
                }]
            },
            timeout=60
        )
        resp.raise_for_status()
        result = resp.json()
        
        if "choices" not in result or not result["choices"]:
            raise ValueError(f"API 返回异常结构: {result}")
            
        data = result["choices"][0]["message"]["content"]
        
        # 混元偶尔输出格式漂移，必须用 Schema 校验兜底
        return ContractSchema.model_validate_json(data).model_dump()
        
    except ValidationError as e:
        # 字段缺失时自动转人工复核队列，不把脏数据写进业务系统
        raise ValueError(f"合同解析格式校验失败: {e}")
    except requests.RequestException as e:
        raise RuntimeError(f"混元 API 调用失败: {e}")

# 测试：解析单份合同
if __name__ == "__main__":
    result = parse_contract("/data/contracts/2026_001.pdf")
    print(result)

import os
import time
import requests
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler

class ContractWatcher(FileSystemEventHandler):
    def on_created(self, event):
        if not event.is_directory and event.src_path.endswith('.pdf'):
            print(f"检测到新文件: {event.src_path}")
            try:
                result = parse_contract(event.src_path)
                # 通过本地 API 触发执行层，支持定时执行和即时触发两种模式
                requests.post(
                    "http://127.0.0.1:8080/api/v1/workflow/trigger",
                    json={
                        "workflow_id": "contract_auto_entry",
                        "payload": result,
                        "trigger_type": "api"
                    },
                    timeout=10
                )
            except Exception as e:
                # 失败时记录日志，避免单份文档错误导致监控中断
                print(f"处理失败: {e}")

def start_monitor():
    observer = Observer()
    observer.schedule(ContractWatcher(), path="/data/incoming_contracts", recursive=False)
    observer.start()
    print("文档监控已启动，按 Ctrl+C 停止...")
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()

if __name__ == "__main__":
    start_monitor()