用 Amazon Textract 增强 Bedrock 知识库：让大体量复杂账单真正可检索

预计阅读时间：11 分钟

把 PDF 直接放进知识库，并不等于文档已经被正确理解。面对跨页表格、扫描图片、多栏排版和字段密集的水电账单，检索质量往往先受制于文本提取，而不是大语言模型本身。

一种更稳妥的处理方式，是把 Amazon Textract 放在 Amazon Bedrock Knowledge Bases 之前：先提取页面文本、表格与表单字段，再将标准化结果写入知识库的数据源。这样，Bedrock 负责切分、向量化和生成回答，Textract 则负责把复杂版面转换成更可靠的机器可读内容。

为什么不能只依赖 PDF 文本层

复杂业务文档通常同时包含几类问题：

扫描件根本没有可用的文本层。
用量、金额和日期分散在表格、页眉、脚注中。
同一字段可能在多个页面重复出现，但含义不同。
多栏布局会让普通解析器打乱阅读顺序。
一个大文件可能包含多个账户或多个计费周期。

如果提取阶段把“本期用量 850 kWh”拆散，或者把上一期金额与本期日期拼到一起，后续嵌入和生成模型很难自行恢复原始关系。因此，知识库定制的重点不只是更换切分参数，而是建立一条可观察的预处理流水线：

PDF 或图片
  -> S3 原始文档区
  -> Textract 异步分析
  -> 页面、表格和键值关系标准化
  -> S3 知识库数据源
  -> Bedrock Knowledge Bases 摄取
  -> Retrieve 或 RetrieveAndGenerate

对于大型 PDF，可以采用 Textract 的异步文档分析流程。账单等结构化文档通常值得启用 TABLES 和 FORMS，因为仅提取连续文本会丢失字段和值之间的关系。

预处理时要保留哪些信息

送入知识库的内容不应只是所有行的简单拼接。至少建议保留以下边界：

页码以及原始文件标识。
账户号、账单周期、服务地址等可过滤字段。
表格行列关系，以及表单键值关系。
金额、单位和日期的原始表示。
Textract 块 ID 或坐标，便于回查原页。

可以把每张账单转换成 Markdown 或纯文本，并用明确标签强化语义。例如：

DOCUMENT_ID: bill-2025-001
ACCOUNT_ID: A-10482
BILLING_PERIOD: 2025-01-01/2025-01-31
PAGE: 1

## Account summary
Previous balance: $84.20
Payments received: $84.20
Current charges: $91.73

## Electricity usage
| Period | Usage | Unit |
| --- | ---: | --- |
| 2025-01 | 850 | kWh |

这种表示方式能让切分后的片段继续携带业务上下文。生产环境还可以生成 Bedrock Knowledge Bases 支持的元数据侧车文件，用账户、地区、文档类型或计费周期限制检索范围，避免相似账单互相污染答案。

可以这样实践：Textract 处理后触发知识库摄取

下面是一个可改造的 Python 示例。它会把本地 PDF 上传到 S3，调用 Textract 异步分析，将页面文本写入知识库数据源目录，然后启动 Bedrock Knowledge Bases 摄取任务。

运行前需要：

安装并配置 AWS CLI 凭证。
为调用身份授予 S3、Textract 和 Bedrock Knowledge Bases 所需权限。
已经创建 Bedrock 知识库及其 S3 数据源。
将环境变量替换成自己的区域、桶、知识库 ID 和数据源 ID。

python -m pip install --upgrade boto3

export AWS_REGION=us-east-1
export SOURCE_BUCKET=my-document-bucket
export KNOWLEDGE_BUCKET=my-knowledge-base-bucket
export KNOWLEDGE_BASE_ID=ABCDEFGHIJ
export DATA_SOURCE_ID=KLMNOPQRST

python ingest_bill.py ./samples/utility-bill.pdf

创建 ingest_bill.py：

import os
import sys
import time
from pathlib import Path

import boto3

REGION = os.environ.get('AWS_REGION', 'us-east-1')
SOURCE_BUCKET = os.environ['SOURCE_BUCKET']
KNOWLEDGE_BUCKET = os.environ['KNOWLEDGE_BUCKET']
KNOWLEDGE_BASE_ID = os.environ['KNOWLEDGE_BASE_ID']
DATA_SOURCE_ID = os.environ['DATA_SOURCE_ID']

s3 = boto3.client('s3', region_name=REGION)
textract = boto3.client('textract', region_name=REGION)
bedrock_agent = boto3.client('bedrock-agent', region_name=REGION)


def wait_for_textract(job_id: str) -> None:
    while True:
        response = textract.get_document_analysis(JobId=job_id, MaxResults=1)
        status = response['JobStatus']
        if status == 'SUCCEEDED':
            return
        if status == 'FAILED':
            raise RuntimeError(response.get('StatusMessage', 'Textract job failed'))
        time.sleep(5)


def collect_lines(job_id: str) -> str:
    pages = {}
    next_token = None

    while True:
        request = {'JobId': job_id, 'MaxResults': 1000}
        if next_token:
            request['NextToken'] = next_token
        response = textract.get_document_analysis(**request)

        for block in response.get('Blocks', []):
            if block.get('BlockType') == 'LINE':
                page = block.get('Page', 1)
                pages.setdefault(page, []).append(block.get('Text', ''))

        next_token = response.get('NextToken')
        if not next_token:
            break

    sections = []
    for page_number in sorted(pages):
        sections.append(f'# Page {page_number}\n')
        sections.extend(pages[page_number])
        sections.append('')
    return '\n'.join(sections)


def wait_for_ingestion(job_id: str) -> None:
    while True:
        response = bedrock_agent.get_ingestion_job(
            knowledgeBaseId=KNOWLEDGE_BASE_ID,
            dataSourceId=DATA_SOURCE_ID,
            ingestionJobId=job_id,
        )
        status = response['ingestionJob']['status']
        print(f'Ingestion status: {status}')
        if status == 'COMPLETE':
            return
        if status in {'FAILED', 'STOPPED'}:
            raise RuntimeError(response['ingestionJob'].get('failureReasons', []))
        time.sleep(10)


def main(pdf_path: str) -> None:
    path = Path(pdf_path)
    source_key = f'raw/{path.name}'
    output_key = f'processed/{path.stem}.txt'

    s3.upload_file(str(path), SOURCE_BUCKET, source_key)
    analysis = textract.start_document_analysis(
        DocumentLocation={
            'S3Object': {'Bucket': SOURCE_BUCKET, 'Name': source_key}
        },
        FeatureTypes=['TABLES', 'FORMS'],
    )

    job_id = analysis['JobId']
    print(f'Textract job: {job_id}')
    wait_for_textract(job_id)

    normalized_text = collect_lines(job_id)
    s3.put_object(
        Bucket=KNOWLEDGE_BUCKET,
        Key=output_key,
        Body=normalized_text.encode('utf-8'),
        ContentType='text/plain; charset=utf-8',
    )
    print(f'Uploaded s3://{KNOWLEDGE_BUCKET}/{output_key}')

    ingestion = bedrock_agent.start_ingestion_job(
        knowledgeBaseId=KNOWLEDGE_BASE_ID,
        dataSourceId=DATA_SOURCE_ID,
        description=f'Ingest {path.name}',
    )
    ingestion_job_id = ingestion['ingestionJob']['ingestionJobId']
    wait_for_ingestion(ingestion_job_id)


if __name__ == '__main__':
    if len(sys.argv) != 2:
        raise SystemExit('Usage: python ingest_bill.py <document.pdf>')
    main(sys.argv[1])

这个最小示例只按页输出 LINE 块，适合验证整条链路。它虽然请求了 TABLES 和 FORMS，但尚未解析 CELL、KEY_VALUE_SET 及其 Relationships。在正式账单场景中，应补充关系解析，将表格重建为 Markdown，将键值对写成稳定的 字段: 值 格式，否则复杂版面的结构信息仍会损失。

查询时同时控制检索范围和回答边界

预处理完成后，可以通过 RetrieveAndGenerate 同时执行检索与回答。下面的示例假设已经设置 MODEL_ARN；具体模型或推理配置应选择当前区域可用的资源。

import os
import boto3

region = os.environ.get('AWS_REGION', 'us-east-1')
client = boto3.client('bedrock-agent-runtime', region_name=region)

response = client.retrieve_and_generate(
    input={
        'text': '账户 A-10482 在 2025 年 1 月用了多少电？本期费用是多少？'
    },
    retrieveAndGenerateConfiguration={
        'type': 'KNOWLEDGE_BASE',
        'knowledgeBaseConfiguration': {
            'knowledgeBaseId': os.environ['KNOWLEDGE_BASE_ID'],
            'modelArn': os.environ['MODEL_ARN'],
            'retrievalConfiguration': {
                'vectorSearchConfiguration': {
                    'numberOfResults': 5
                }
            },
            'generationConfiguration': {
                'promptTemplate': {
                    'textPromptTemplate': (
                        '仅依据以下检索结果回答问题。'
                        '金额、日期、账户和计量单位必须与原文一致。'
                        '若证据不足，请明确回答无法确定。\n\n'
                        '$search_results$\n\n问题：$query$'
                    )
                }
            },
        },
    },
)

print(response['output']['text'])
for citation in response.get('citations', []):
    print(citation.get('retrievedReferences', []))

对于客户服务场景，不应只展示生成文本。应用还应保存引用片段、原文件位置和请求追踪 ID，让客服人员能够核对金额与日期。账户号等敏感字段需要结合 IAM、加密、日志脱敏和租户级过滤进行保护。

上线前的检查清单

评估这套方案时，不要只测试“能不能回答”，还要测量字段级正确率和检索稳定性：

准备不同供应商、扫描质量、页数和版式的代表性账单集。
分别核验账户号、日期、用量、单位、税费和总金额。
检查跨页表格是否在切分后仍保留表头。
对比纯文本解析与 Textract 预处理后的召回率。
记录 Textract、嵌入、向量存储和模型调用的延迟与成本。
为重复上传设计幂等键，避免同一账单被多次摄取。
对低置信度字段设置人工复核流程，不让模型猜测关键金额。

Textract 增加了一段处理流程，也带来了额外成本和延迟；它更适合扫描件多、版面复杂、字段准确性要求高的知识库。对于文本层干净、结构简单的 PDF，直接摄取可能已经足够。实际落地时，应先用一组真实文档做基准测试，再决定哪些文档走 Textract 增强链路，哪些文档保留轻量处理路径。