仅需几行代码，自动化贷款承销流水线

一份典型的贷款文件就是一堆工资单和券商对账单，每份格式都不同，每个数字都得手动重新录入。以下是一个使用 LlamaParse 自动完成的流水线：PDF 转为干净 markdown，字段提取到 Pydantic 模型，然后进行跨文档分析，生成带有差异标记的承销摘要。全文和仓库链接：https://llamaindex.ai/blog/building-a-financial-document-pipeline-with-llamaparse…

使用 LlamaParse 构建贷款承销流水线

来源：https://www.llamaindex.ai/blog/building-a-financial-document-pipeline-with-llamaparse

贷款承销需要从多份财务文档中提取数据。这通常包括工资单和券商对账单，它们的布局复杂且不同供应商之间差异很大。这是一个关键的金融工作流程，往往需要大量人工核查和重复流程。

上周我（Logan，LlamaIndex 开源主管）在纽约举办了一场实践工作坊，开发者们使用 LlamaParse 工具从头构建了一个贷款承销流水线。最终的应用能够处理杂乱的金融 PDF（工资单、券商对账单），提取结构化数据，并进行跨文档分析。

我们希望构建的流水线能够：

1. 解析 PDF 为干净的 markdown，使用 LlamaParse 的 agentic 层级
2. 提取 结构化字段（雇主名称、总工资、持仓、账户价值）到类型化的 Pydantic 模型
3. 分析 跨文档数据，生成带有差异标记的承销摘要
4. 审核 分析结果，并引入人工审批环节

这篇文章将介绍我们构建的内容以及如何亲自尝试。

工作坊技术栈

为了适应工作坊环境，技术栈故意保持简单。我们结合使用了异步 Python、SQLite、FastAPI、Pydantic 和 LlamaCloud SDK，构建了一个完全异步的流水线，并配有内存任务队列。虽然技术栈简单，但架构设计具有可扩展性。你（或你的编码助手）可以轻松根据需要替换组件。例如，可以用 Celery 或 Temporal 替换任务队列，用 Postgres 替换数据库，用 S3 替换本地文件存储。

LlamaParse 的三种用法

工作坊让参与者实现了三个服务文件，每个文件以不同方式使用 LlamaParse。

1. 解析：PDF 转 Markdown

第一个服务上传 PDF 并返回干净的 markdown。这是 LlamaParse 的核心能力，其 agentic 解析层级能处理不同薪资提供商和券商之间杂乱的表格布局和格式不一致问题。

import asyncio
from llama_cloud import AsyncLlamaCloud

client = AsyncLlamaCloud(api_key=settings.llama_cloud_api_key)
file_obj = await client.files.create(file=file_path, purpose="parse")
job = await client.parsing.create(file_id=file_obj.id, tier="agentic", version="latest")

# 轮询直到完成
result = await client.parsing.get(job.id, expand=["markdown_full"])
while result.job.status not in ("COMPLETED", "FAILED", "CANCELLED"):
    await asyncio.sleep(3)
    result = await client.parsing.get(job.id, expand=["markdown_full"])
parsed_markdown = result.markdown_full

三个 API 调用：上传文件、创建任务、轮询结果。返回的 markdown 保留了表格结构，这对下一步至关重要。

2. 提取：结构化字段

第二个服务接收解析后的文档，并使用 Pydantic schema 提取类型化字段。你定义需要提取的内容，LlamaParse 负责提取。例如，工资单 schema：

from pydantic import BaseModel

class PayStub(BaseModel):
    employer_name: str
    employee_name: str
    pay_period_start: str
    pay_period_end: str
    gross_pay: float
    net_pay: float
    ytd_gross_income: float
    deductions: list[Deduction]

提取调用将 schema 作为 JSON Schema 传递给 LlamaParse：

job = await client.extract.create(
    file_input=file_obj.id,
    configuration={
        "tier": "agentic",
        "data_schema": PayStub.model_json_schema(),
    }
)

与解析类似，上传文件，使用文件 ID，然后轮询结果。任务完成后，可以使用 PayStub.model_validate(result.extract_result) 根据 schema 进行验证。

3. 跨文档分析

第三个服务最为有趣。它接收来自多个文档的已提取数据（例如，一份工资单和一份券商对账单），将它们合并到一个文本缓冲区中，将该缓冲区上传到 LlamaParse，再次运行提取。但这次，使用的是跨越所有文档的承销摘要 schema，它执行更多推理而非纯提取。

# 将提取的数据合并为一个文本文档
text = _format_extractions_as_text(extracted_data)

# 作为缓冲区文件上传
file_obj = await client.files.create(
    file=(f"review_{review_id}.txt", io.BytesIO(text.encode("utf-8"))),
    purpose="extract",
)

# 使用跨文档 schema 进行提取
job = await client.extract.create(
    file_input=file_obj.id,
    configuration={
        "tier": "agentic",
        "data_schema": UnderwritingSummary.model_json_schema(),
        "system_prompt": "You are a loan underwriter. Analyze ...",
    }
)

承销摘要 schema 要求提供已验证的收入、总流动资产、储备金月数，以及带严重性评级的差异列表。通过设置系统提示词，我们可以明确提示服务执行跨文档分析，而不仅仅是提取字段。这正是可以将业务特定知识注入流水线，从而产生更具可操作性的输出的地方。

亲自尝试

首先，如果你还没有 LlamaCloud（https://cloud.llamaindex.ai/）的 API 密钥，请先获取一个。仓库已配置好，你可以逐步实现每个服务。每个阶段都有一个分支，包含了 TODO 存根：

开始：

git clone && cd finparse-pipeline
git checkout phase_1 # 从 TODO 存根开始
uv sync --group dev
cp .env.example .env   # 添加你的 LLAMA_CLOUD_API_KEY
uv run fastapi dev app/main.py

在 http://localhost:8000/docs 的 Swagger UI 可以驱动整个流程：上传 PDF、轮询解析、触发提取、创建审核。