用 Pulse AI + Amazon Bedrock 搭建金融文档抽取与微调流水线

PDF 文档
  │
  ▼
Pulse AI 文档理解服务
  │  → 输出：结构化 JSON（表格、段落、键值对、跨页链接）
  ▼
预处理脚本（清洗、字段映射、对齐）
  │
  ▼
Amazon Bedrock
  │  ├─ 推理：用 Claude / Titan 等模型做语义校验与补全
  │  ├─ 微调：用抽取结果构建训练集，定制化模型
  ▼
下游系统（数据库、风控引擎、报表平台）

import requests
import json
import time

PULSE_API_URL = "https://api.pulseai.example.com/v1/documents"
PULSE_API_KEY = "your-pulse-api-key"  # 替换为真实密钥

def extract_financial_document(pdf_path: str) -> dict:
    """上传 PDF 到 Pulse AI，等待结构化抽取完成，返回 JSON 结果。"""

    # 1. 上传文档
    with open(pdf_path, "rb") as f:
        upload_resp = requests.post(
            f"{PULSE_API_URL}/upload",
            headers={"Authorization": f"Bearer {PULSE_API_KEY}"},
            files={"file": (pdf_path, f, "application/pdf")},
            data={"doc_type": "annual_report"}  # 指定文档类型，提升抽取精度
        )
    upload_resp.raise_for_status()
    doc_id = upload_resp.json()["document_id"]

    # 2. 等待处理完成（轮询，生产环境建议用 webhook）
    for _ in range(60):
        status_resp = requests.get(
            f"{PULSE_API_URL}/{doc_id}/status",
            headers={"Authorization": f"Bearer {PULSE_API_KEY}"},
        )
        status = status_resp.json()["status"]
        if status == "completed":
            break
        if status == "failed":
            raise RuntimeError(f"Pulse AI 处理失败: {status_resp.json()['error']}")
        time.sleep(5)

    # 3. 获取结构化结果
    result_resp = requests.get(
        f"{PULSE_API_URL}/{doc_id}/result",
        headers={"Authorization": f"Bearer {PULSE_API_KEY}"},
    )
    result_resp.raise_for_status()
    return result_resp.json()


# 使用示例
result = extract_financial_document("annual_report_2024.pdf")

# result 的典型结构（示意）：
# {
#   "tables": [
#     {
#       "type": "balance_sheet",
#       "page": 12,
#       "rows": [
#         {"label": "资产总计", "merged": "5,832,400", "parent_only": "3,210,800"},
#         ...
#       ]
#     }
#   ],
#   "key_value_pairs": [
#     {"key": "审计意见", "value": "无保留意见", "page": 87}
#   ],
#   "cross_references": [
#     {"from_page": 12, "from_text": "详见附注15", "to_page": 87, "to_section": "应收账款"}
#   ]
# }

print(json.dumps(result, indent=2, ensure_ascii=False))

import boto3
import json

bedrock = boto3.client("bedrock-runtime", region_name="us-east-1")

MODEL_ID = "anthropic.claude-3-5-sonnet-20241022-v2:0"

def validate_balance_sheet(extracted_tables: list) -> dict:
    """用 Bedrock Claude 校验资产负债表抽取结果的逻辑一致性。"""

    # 找到资产负债表
    bs_table = next(t for t in extracted_tables if t["type"] == "balance_sheet")

    # 构造 prompt——把结构化数据塞进去，让模型做校验
    prompt = f"""你是一名资深财务分析师。以下是从年报中抽取的资产负债表结构化数据：

{json.dumps(bs_table["rows"], indent=2, ensure_ascii=False)}

请逐项检查：
1. 资产总计 = 流动资产合计 + 非流动资产合计？
2. 负债总计 = 流动负债合计 + 非流动负债合计？
3. 资产总计 = 负债总计 + 所有者权益合计？
4. 合并列与母公司列之间的差异是否合理？

如果发现不一致，指出具体行和数值。如果一致，给出确认结论。"""

    body = {
        "anthropic_version": "bedrock-2023-05-31",
        "max_tokens": 2048,
        "messages": [
            {"role": "user", "content": prompt}
        ]
    }

    response = bedrock.invoke_model(
        modelId=MODEL_ID,
        body=json.dumps(body),
        contentType="application/json",
        accept="application/json",
    )

    result = json.loads(response["body"].read())
    return {
        "validation_result": result["content"][0]["text"],
        "input_tokens": result["usage"]["input_tokens"],
        "output_tokens": result["usage"]["output_tokens"],
    }


# 使用示例
validation = validate_balance_sheet(result["tables"])
print(validation["validation_result"])
print(f"Token 消耗: 输入 {validation['input_tokens']}, 输出 {validation['output_tokens']}")

import boto3
import json

bedrock = boto3.client("bedrock", region_name="us-east-1")

# 1. 构造微调数据集（从已校验的抽取结果中生成）
def build_finetune_dataset(extracted_results: list[dict]) -> list[dict]:
    """把多份文档的抽取+校验结果转成 Bedrock 微调格式。"""

    training_examples = []
    for doc_result in extracted_results:
        # 每份文档生成多条 instruction-following 样本
        for table in doc_result["tables"]:
            example = {
                "system": "你是一名金融文档分析专家，擅长从结构化数据中提取和校验财务指标。",
                "messages": [
                    {
                        "role": "user",
                        "content": f"请从以下结构化表格中提取关键财务指标并校验一致性：\n{json.dumps(table['rows'], ensure_ascii=False)}"
                    },
                    {
                        "role": "assistant",
                        # 这里填入人工校验后的正确输出
                        "content": table["validated_summary"]  # 需提前标注
                    }
                ]
            }
            training_examples.append(example)

    return training_examples


# 2. 上传训练集到 S3
s3 = boto3.client("s3")
BUCKET = "my-financial-ml-datasets"

dataset = build_finetune_dataset(verified_results)  # verified_results 是已标注数据

s3.put_object(
    Bucket=BUCKET,
    Key="finetune/financial_extraction_train.jsonl",
    Body="\n".join(json.dumps(ex, ensure_ascii=False) for ex in dataset),
)

# 3. 创建微调任务
training_job = bedrock.create_model_customization_job(
    jobName="financial-doc-finetune-v1",
    customModelName="financial-doc-custom-v1",
    baseModelIdentifier="anthropic.claude-3-5-sonnet-20241022-v2:0",
    trainingDataConfig={
        "s3Uri": f"s3://{BUCKET}/finetune/financial_extraction_train.jsonl"
    },
    outputDataConfig={
        "s3Uri": f"s3://{BUCKET}/finetune/output/"
    },
    hyperParameters={
        "epochCount": "3",
        "learningRate": "0.0001",
        "batchSize": "4",
    },
    roleArn="arn:aws:iam::123456789012:role/BedrockFineTuneRole",  # 替换为你的 IAM Role
)

print(f"微调任务已创建: {training_job['jobArn']}")