Python sdk - Docflow API

使用 docflow-sdk Python SDK 快速集成 Docflow 文档流程管理能力

docflow-sdk 是 TextIn Docflow 的官方 Python SDK，提供工作空间管理、文档分类、智能审核、类型安全的响应模型和完善的错误处理。

PyPI: docflow-sdk
GitHub: docflow-python-sdk

安装

pip install docflow-sdk

git clone https://github.com/intsig-textin/docflow-python-sdk.git
cd docflow-python-sdk
pip install -e .

系统要求： Python >= 3.8

认证与初始化

SDK 支持多种认证方式（优先级：构造参数 > 环境变量 > .env 文件）：

import os
os.environ["DOCFLOW_APP_ID"] = "your-app-id"
os.environ["DOCFLOW_SECRET_CODE"] = "your-secret-code"
os.environ["DOCFLOW_BASE_URL"] = "https://docflow.textin.com/api"

from docflow import DocflowClient
client = DocflowClient.from_env()

from docflow import DocflowClient
client = DocflowClient(
    app_id="your-app-id",
    secret_code="your-secret-code",
    base_url="https://docflow.textin.com/api",
)

# 需要安装 dotenv: pip install python-dotenv
# .env 文件内容:
# DOCFLOW_APP_ID=your-app-id
# DOCFLOW_SECRET_CODE=your-secret-code
# DOCFLOW_BASE_URL=https://docflow.textin.com/api

from dotenv import load_dotenv
from docflow import DocflowClient

load_dotenv()
client = DocflowClient.from_env()

推荐使用环境变量方式，避免在代码中硬编码密钥。

API 概览

资源	说明	主要方法
`client.workspace`	工作空间管理	`create()`, `list()`, `get()`, `update()`, `delete()`, `iter()`
`client.category`	文档类别管理	`create()`, `list()`, `get()`, `update()`, `delete()`, `iter()`
`client.category.fields`	类别字段管理	`batch_add()`, `list()`, `batch_update()`, `delete()`
`client.category.tables`	类别表格管理	`batch_add()`, `list()`, `batch_update()`, `delete()`
`client.category.samples`	类别样本管理	`batch_upload()`, `list()`, `download()`, `delete()`
`client.file`	文件管理与识别	`upload()`, `fetch()`, `download()`, `delete()`
`client.review`	智能审核规则管理	`create_repo()`, `create_group()`, `create_rule()`, `submit_task()`

工作空间管理

工作空间是 Docflow 的顶层组织单位，用于隔离不同业务场景的文档处理流程。

创建工作空间

from docflow import DocflowClient, AuthScope

client = DocflowClient.from_env()

workspace = client.workspace.create(
    name="费用报销处理空间",
    auth_scope=AuthScope.PUBLIC,  # 公开权限
    description="处理公司费用报销单据"
)

print(f"工作空间ID: {workspace.workspace_id}")

获取工作空间列表

# 分页获取
workspaces = client.workspace.list(page=1, page_size=20)

for ws in workspaces.workspaces:
    print(f"{ws.workspace_id}: {ws.name}")

# 使用迭代器自动分页
for workspace in client.workspace.iter():
    print(f"{workspace.workspace_id}: {workspace.name}")

链式调用（推荐）

通过上下文绑定简化代码，减少重复参数传递：

# 绑定工作空间
ws = client.workspace("123")

# 获取工作空间详情
detail = ws.get()

# 更新工作空间
ws.update(
    name="新的工作空间名称",
    auth_scope=AuthScope.PRIVATE
)

# 链式调用：操作类别
cat = ws.category("456")
cat.fields.batch_add(fields=[{"name": "发票号码"}])
cat.tables.batch_add(tables=[{"name": "商品明细表"}])

文档类别管理

类别定义文档的结构化字段和抽取规则。

创建类别（带样本文件）

from docflow import ExtractModel, FieldType

# 定义字段配置
fields = [
    {
        "name": "发票号码",
        "description": "发票唯一标识"
    },
    {
        "name": "开票日期",
        "transform_settings": {
            "type": FieldType.DATETIME.value,
            "datetime_settings": {
                "format": "yyyy-MM-dd"
            }
        }
    },
    {
        "name": "发票类型",
        "transform_settings": {
            "type": FieldType.ENUMERATE.value,
            "enumerate_settings": {
                "items": ["增值税专用发票", "增值税普通发票"]
            }
        }
    }
]

# 创建类别
category = client.category.create(
    workspace_id="123",
    name="增值税发票",
    extract_model=ExtractModel.Model_1,  # 使用模型1
    sample_files=[
        "/path/to/invoice_sample1.pdf",
        "/path/to/invoice_sample2.pdf"
    ],
    fields=fields,
    category_prompt="增值税发票，包含发票信息和商品明细"
)

print(f"类别ID: {category.category_id}")

# 创建类别后，添加表格
cat = client.workspace("123").category(category.category_id)
tables = cat.tables.batch_add(tables=[{
    "name": "商品明细表",
    "prompt": "抽取商品名称、规格、数量、单价、金额"
}])
print(f"表格ID: {tables[0].table_id}")

分类关键词规则

创建或更新类别时，可通过 category_keyword_rules 配置分类关键词规则。正向规则组之间为 OR；命中反向规则组时排除该类别。 SDK 默认会分页读取当前工作空间内已启用类别，并在请求前预检同类型规则组的跨类别冲突；可传 check_keyword_rule_conflicts=False 关闭该预检以减少请求。服务端仍会执行最终冲突校验。

rules = {
    "positive_rules": [{
        "group_name": "发票核心要素",
        "min_hit": 2,
        "words": ["发票号码", "购买方名称", "价税合计"],
    }],
    "negative_rules": [{
        "group_name": "排除收据",
        "min_hit": 1,
        "words": ["收据", "付款收据"],
    }],
}

client.category.create(
    workspace_id="123", name="增值税发票", extract_model=ExtractModel.Model_1,
    sample_files=["/path/to/invoice.pdf"], fields=[{"name": "发票号码"}],
    category_keyword_rules=rules,
)

# 更新时传入双空数组，显式清空已配置规则
client.category.update(
    workspace_id="123", category_id="456",
    category_keyword_rules={"positive_rules": [], "negative_rules": []},
)

创建时不传该参数表示不配置规则；传入双空数组也等价于不配置。
更新时不传或传 None 表示不修改；传有效对象会全量替换；仅双空数组表示清空。{} 和空白字符串无效。
SDK 中规则组的 group_name 可省略：正向组会按顺序自动补为 group_1、group_2……，反向组补为 exclude。它仅用于展示/追踪，不参与分类命中或冲突判断；从接口读取后再更新时会保留服务端返回的名称。
关键词保留输入时的大小写和空格。每个关键词不得为空或超过 50 个字符，同一规则组内不得包含原始字符串完全相同的关键词；每组最多 20 个关键词，最多 10 个正向组和 1 个反向组；min_hit 必须在 1 到该组关键词数之间。
同一类别的正反向规则不能包含原始字符串完全相同的关键词；同空间其他已启用类别不能有原始关键词集合相同的同类型规则组。SDK 会预校验可在本地判断的规则，服务端仍会做最终校验；跨类别冲突返回错误码 1005015019。

字段管理

# 绑定类别上下文
cat = client.workspace("123").category("456")

# 新增字段
result = cat.fields.batch_add(fields=[
    {"name": "税率", "description": "税率百分比"}
])

# 获取字段列表
fields = cat.fields.list()
for field in fields.fields:
    print(f"{field.id}: {field.name}")

# 更新字段
cat.fields.batch_update(fields=[
    {"field_id": "789", "name": "税率(%)", "required": True}
])

# 删除字段
cat.fields.delete(field_ids=["789"])

表格管理

# 添加表格（支持内嵌字段一站式创建）
tables = cat.tables.batch_add(tables=[{
    "name": "商品明细表",
    "prompt": "抽取商品名称、规格、数量、单价、金额",
    "fields": [
        {"name": "商品名称"},
        {"name": "数量"},
        {"name": "单价"}
    ]
}])

# 获取表格列表
tables = cat.tables.list()
for table in tables.tables:
    print(f"{table.table_id}: {table.name}")

样本管理

# 上传样本文件（支持多文件）
result = cat.samples.batch_upload(files=[
    "/path/to/invoice_sample1.pdf",
    "/path/to/invoice_sample2.pdf"
])

# 获取样本列表
samples = cat.samples.list()

# 下载样本（ZIP 打包）
cat.samples.batch_download(
    sample_ids=["789"],
    save_path="/path/to/samples.zip"
)

# 删除样本
cat.samples.delete(sample_ids=["789"])

文件处理

上传文件并识别

# 上传文件（自动开始识别）
response = client.file.upload(
    workspace_id="123",
    category="增值税发票",  # 类别名称
    file_path="/path/to/invoice.pdf"
)

batch_number = response.batch_number
print(f"批次号: {batch_number}")

获取识别结果

import time

# 轮询等待识别完成
max_wait = 60
wait_interval = 3
elapsed = 0

while elapsed < max_wait:
    result = client.file.fetch(
        workspace_id="123",
        batch_number=batch_number
    )
    
    if result.files and result.files[0].recognition_status == 1:
        file_info = result.files[0]
        print(f"识别完成: {file_info.name}")
        
        # 访问抽取的字段数据
        if file_info.data and 'fields' in file_info.data:
            for field in file_info.data['fields']:
                print(f"{field['name']}: {field['value']}")
        
        # 访问表格数据
        if file_info.data and 'items' in file_info.data:
            for row in file_info.data['items']:
                print(row)
        
        break
    
    time.sleep(wait_interval)
    elapsed += wait_interval

智能审核

Docflow 提供基于 LLM 的智能审核能力，支持单文档规则校验和跨文档交叉审核。

创建审核规则库

# 1. 创建规则库
repo = client.review.create_repo(
    workspace_id="123",
    name="费用报销审核规则库"
)

# 2. 创建规则组
group = client.review.create_group(
    workspace_id="123",
    repo_id=repo.repo_id,
    name="发票合规性检查"
)

# 3. 创建审核规则
# 获取字段ID映射
fields_response = client.category.fields.list(
    workspace_id="123",
    category_id="456"
)
field_map = {f.name: f.id for f in fields_response.fields}

# 创建规则：必填字段完整性校验
client.review.create_rule(
    workspace_id="123",
    repo_id=int(repo.repo_id),
    group_id=group.group_id,
    name="必填字段完整性校验",
    prompt='检查"发票号码"、"开票日期"、"金额"是否都已填写，任一字段为空则审核不通过',
    category_ids=["456"],
    risk_level=10,  # 高风险
    referenced_fields=[
        {
            "category_id": "456",
            "category_name": "增值税发票",
            "fields": [
                {"field_id": field_map["发票号码"], "field_name": "发票号码"},
                {"field_id": field_map["开票日期"], "field_name": "开票日期"},
                {"field_id": field_map["金额"], "field_name": "金额"}
            ],
            "tables": []
        }
    ]
)

跨文档交叉审核

# 创建跨文档规则：验证发票金额与付款金额一致
client.review.create_rule(
    workspace_id="123",
    repo_id=int(repo.repo_id),
    group_id=group.group_id,
    name="跨文档金额匹配",
    prompt="验证发票金额与支付记录交易金额一致，允许±0.1元误差",
    category_ids=["456", "789"],  # 多个类别
    risk_level=10,
    referenced_fields=[
        {
            "category_id": "456",
            "category_name": "增值税发票",
            "fields": [
                {"field_id": invoice_field_map["金额"], "field_name": "金额"}
            ],
            "tables": []
        },
        {
            "category_id": "789",
            "category_name": "支付记录",
            "fields": [
                {"field_id": payment_field_map["交易金额"], "field_name": "交易金额"}
            ],
            "tables": []
        }
    ]
)

提交审核任务

# 提交审核任务
review_task = client.review.submit_task(
    workspace_id="123",
    name="2024年3月费用报销审核",
    repo_id=repo.repo_id,
    extract_task_ids=["task_001", "task_002", "task_003"]
)

task_id = review_task['task_id']
print(f"审核任务ID: {task_id}")

获取审核结果

import time

# 轮询等待审核完成
max_wait = 120
wait_interval = 5
elapsed = 0

while elapsed < max_wait:
    result = client.review.get_task_result(
        workspace_id="123",
        task_id=task_id
    )
    
    status = result.get('status')
    # 状态：0=待审核, 1=审核通过, 2=审核失败, 4=审核不通过, 7=识别失败
    if status in (1, 2, 4, 7):
        print("审核完成")
        
        # 输出审核结果
        stats = result.get('statistics', {})
        print(f"规则通过数: {stats.get('pass_count', 0)}")
        print(f"规则不通过数: {stats.get('failure_count', 0)}")
        
        # 详细审核结果
        for group in result.get('groups', []):
            print(f"\n【{group['group_name']}】")
            for task in group.get('review_tasks', []):
                result_icon = "✓" if task['review_result'] == 0 else "✗"
                print(f"  {result_icon} {task['rule_name']}")
                print(f"    {task['reasoning']}")
        
        break
    
    time.sleep(wait_interval)
    elapsed += wait_interval

枚举类型

SDK 提供完整的枚举类型定义，避免参数传错：

from docflow import (
    ExtractModel,      # 提取模型类型
    EnabledStatus,     # 启用状态（查询用）
    EnabledFlag,       # 启用标志（更新用）
    AuthScope,         # 权限范围
    FieldType,         # 字段类型
    MismatchAction,    # 不匹配处理模式
    RecognitionStatus, # 识别状态
)

# ExtractModel - 提取模型
ExtractModel.Model_1  # 速度快，抽取结果稳定
ExtractModel.Model_2  # 适用复杂文档理解
ExtractModel.Model_3  # 多模态，适用简单抽取

# AuthScope - 权限范围
AuthScope.PRIVATE  # 0 - 私有权限
AuthScope.PUBLIC   # 1 - 公共权限

# EnabledStatus - 启用状态（查询用）
EnabledStatus.ALL       # "all" - 全部
EnabledStatus.DISABLED  # "0"   - 未启用
EnabledStatus.ENABLED   # "1"   - 已启用

# EnabledFlag - 启用标志（更新用）
EnabledFlag.DISABLED  # 0 - 未启用
EnabledFlag.ENABLED   # 1 - 已启用

# FieldType - 字段转换类型
FieldType.DATETIME   # "datetime"   - 日期时间
FieldType.ENUMERATE  # "enumerate"  - 枚举
FieldType.REGEX      # "regex"      - 正则表达式

# RecognitionStatus - 识别状态
RecognitionStatus.PENDING    # 0 - 待识别
RecognitionStatus.SUCCESS    # 1 - 识别成功
RecognitionStatus.FAILED     # 2 - 识别失败

自动分页迭代器

使用迭代器自动处理分页，无需手动循环：

# 工作空间迭代器
for workspace in client.workspace.iter():
    print(f"{workspace.workspace_id}: {workspace.name}")
    if some_condition:
        break  # 可以随时中断

# 类别迭代器
for category in client.category.iter(workspace_id="123"):
    print(f"{category.category_id}: {category.name}")

# 限制最大页数
for category in client.category.iter(workspace_id="123", max_pages=5):
    print(category.name)

# 转换为列表（获取所有数据）
all_workspaces = list(client.workspace.iter())

错误处理

SDK 提供了完善的错误分类，方便精确处理不同的异常情况。

错误类型

错误类	说明
`DocflowException`	基础错误类，捕获所有 SDK 错误
`ValidationError`	参数校验失败
`AuthenticationError`	认证失败（app-id 或 secret-code 错误）
`PermissionDeniedError`	权限不足
`ResourceNotFoundError`	资源不存在
`APIError`	API 调用失败（HTTP 4xx/5xx）
`NetworkError`	网络连接错误

错误处理示例

from docflow.exceptions import (
    DocflowException,
    AuthenticationError,
    ValidationError,
    ResourceNotFoundError,
    APIError,
)

try:
    workspace = client.workspace.get(workspace_id="123")
except AuthenticationError as e:
    print(f"认证失败: {e.message}")
except ResourceNotFoundError as e:
    print(f"工作空间不存在: {e.message}")
except ValidationError as e:
    print(f"参数校验失败: {e.message}")
except APIError as e:
    print(f"API错误 [HTTP {e.status_code}]: {e.message}")
except DocflowException as e:
    print(f"SDK错误: {e.message}")

国际化（i18n）

SDK 支持多语言错误消息：

from docflow import DocflowClient, set_language

# 使用英文
set_language('en_US')
client = DocflowClient.from_env()

# 使用中文（默认）
set_language('zh_CN')

# 动态切换语言
client.set_language('en_US')

# 获取当前语言
current_lang = client.get_language()  # 'en_US' 或 'zh_CN'

高级配置

超时与重试

client = DocflowClient(
    app_id="your-app-id",
    secret_code="your-secret-code",
    timeout=60,           # 请求超时（秒），默认 30
    max_retries=5,        # 最大重试次数，默认 3
    retry_backoff_factor=1.0,  # 退避因子，默认 1.0
)

自定义重试配置

# 自定义重试状态码
client = DocflowClient(
    app_id="your-app-id",
    secret_code="your-secret-code",
    retry_status_codes=[429, 503],  # 只重试 429 和 503
)

# 自定义重试方法
client = DocflowClient(
    app_id="your-app-id",
    secret_code="your-secret-code",
    retry_methods=["GET"],  # 只允许 GET 请求重试
)

# 禁用重试
client = DocflowClient(
    app_id="your-app-id",
    secret_code="your-secret-code",
    max_retries=0  # 禁用重试
)

自定义 API 地址

client = DocflowClient(
    app_id="your-app-id",
    secret_code="your-secret-code",
    base_url="https://custom-api.example.com"
)

资源管理

使用上下文管理器自动关闭连接：

with DocflowClient.from_env() as client:
    workspaces = client.workspace.list()
    # 退出时自动关闭连接

调试日志

启用 DEBUG 级别日志查看请求详情：

import logging
logging.getLogger("docflow").setLevel(logging.DEBUG)

完整示例

查看 examples 目录获取完整的使用示例：

快速开始: quick_start.py - 费用报销场景端到端流程
完整工作流: complete_workflow_example.py - 发票处理完整流程
文件处理: file_examples.py - 文件上传、识别、下载示例
审核规则: review_examples.py - 审核规则配置示例

常见问题

问题	解决方案
`AuthenticationError`	检查 `DOCFLOW_APP_ID` 和 `DOCFLOW_SECRET_CODE` 是否正确
`ResourceNotFoundError`	确认工作空间ID/类别ID是否存在且有权限访问
`ValidationError`	检查参数格式和取值范围（如工作空间名称最大50字符）
识别结果为空	确认类别配置正确（字段、表格、样本文件），等待识别完成后再获取结果
审核任务失败	检查规则配置（referenced_fields 的 field_id 必须正确）和 extract_task_ids 有效性

​安装

​认证与初始化

​API 概览

​工作空间管理

​创建工作空间

​获取工作空间列表

​链式调用（推荐）

​文档类别管理

​创建类别（带样本文件）

​分类关键词规则

​字段管理

​表格管理

​样本管理

​文件处理

​上传文件并识别

​获取识别结果

​智能审核

​创建审核规则库

​跨文档交叉审核

​提交审核任务

​获取审核结果

​枚举类型

​自动分页迭代器

​错误处理

​错误类型

​错误处理示例

​国际化（i18n）

​高级配置

​超时与重试

​自定义重试配置

​自定义 API 地址

​资源管理

​调试日志

​完整示例

​常见问题

​相关链接

安装

认证与初始化

API 概览

工作空间管理

创建工作空间

获取工作空间列表

链式调用（推荐）

文档类别管理

创建类别（带样本文件）

分类关键词规则

字段管理

表格管理

样本管理

文件处理

上传文件并识别

获取识别结果

智能审核

创建审核规则库

跨文档交叉审核

提交审核任务

获取审核结果

枚举类型

自动分页迭代器

错误处理

错误类型

错误处理示例

国际化（i18n）

高级配置

超时与重试

自定义重试配置

自定义 API 地址

资源管理

调试日志

完整示例

常见问题

相关链接