Docflow Python SDK

简体中文 | English

Docflow Python SDK，提供简洁易用的 API 接口，方便 Python 开发者快速集成Docflow的功能。

📋 功能特性

• ✅ 工作空间管理：创建、查询、更新、删除工作空间
• ✅ 类别管理：创建、查询、更新、删除文件类别（支持文件上传）
• ✅ 字段管理：类别字段的增删改查及配置管理，支持批量新增/更新
• ✅ 表格管理：类别表格的增删改查，支持批量新增/更新（含内嵌字段）
• ✅ 样本管理：样本文件的上传、下载、删除，支持批量上传/批量下载（ZIP）
• ✅ 类型安全：完整的类型注解，支持 IDE 自动补全
• ✅ 枚举类型：提供完整的枚举定义，避免参数传错
• ✅ 异常处理：完善的异常体系，便于错误处理
• ✅ 国际化（i18n）：多语言错误消息支持
  • 支持中文（zh_CN）和英文（en_US）
  • 可动态切换语言
  • 支持参数化消息
• ✅ 智能重试：基于 urllib3.util.retry.Retry 的灵活重试机制
  • 可配置重试状态码（默认：423, 429, 500, 503, 504, 900）
  • 可配置重试方法（默认：GET, POST, PUT, DELETE, PATCH）
  • 可配置退避因子（默认：1.0，指数退避策略）
  • 可配置最大重试次数（默认：3次）
• ✅ 连接池：高效的 HTTP 连接复用（10 连接池，20 最大连接）
• ✅ 易于使用：符合 Python 习惯的 API 设计
• ✅ 链式调用：支持上下文绑定，减少重复参数传递（代码量减少 70%）
• ✅ 自动分页：提供迭代器自动处理分页逻辑，内存高效
• ✅ 环境变量支持：from_env() 方法自动加载配置

📦 安装

使用 pip 安装（推荐）

pip install docflow-sdk

从 GitHub 安装

pip install git+https://github.com/intsig-textin/docflow-python-sdk.git

从源码安装

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

🚀 快速开始

💡 想快速体验完整流程？ 查看 费用报销场景快速开始指南，5分钟完成从创建空间到智能审核的端到端流程！

1. 初始化客户端

from docflow import DocflowClient, AuthScope  # 导入枚举类型

# 方式 1: 最简洁（使用默认 base_url）
client = DocflowClient(
    app_id="your-app-id",
    secret_code="your-secret-code"
    # base_url 默认为 "https://docflow.textin.com/api"
)

# 方式 2: 从环境变量加载
client = DocflowClient.from_env()
# 需要设置环境变量：
# export DOCFLOW_APP_ID="your-app-id"
# export DOCFLOW_SECRET_CODE="your-secret-code"
# export DOCFLOW_BASE_URL="https://docflow.textin.com/api"  # 可选

# 方式 3: 自定义 base_url
client = DocflowClient(
    app_id="your-app-id",
    secret_code="your-secret-code",
    base_url="https://custom.api.com/api"
)

2. 工作空间操作

创建工作空间

# 使用 AuthScope 枚举避免参数传错
workspace = client.workspace.create(
    name="我的工作空间",
    auth_scope=AuthScope.PUBLIC  # 使用枚举而非数字1
)
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}")

获取工作空间详情

detail = client.workspace.get(workspace_id="123")
print(f"工作空间名称: {detail.name}")
print(f"权限范围: {detail.auth_scope}")
print(f"管理员ID: {detail.manage_account_id}")

更新工作空间

client.workspace.update(
    workspace_id="123",
    name="新的工作空间名称",
    auth_scope=AuthScope.PRIVATE
)

删除工作空间

client.workspace.delete(workspace_ids=["123", "456"])

3. 类别操作

创建类别（带文件上传）

from docflow import ExtractModel, FieldType  # 导入枚举类型

# 定义字段配置
fields = [
    {
        "name": "发票号码",
        "description": "发票唯一标识"
    },
    {
        "name": "开票日期",
        "transform_settings": {
            "type": FieldType.DATETIME.value,  # 使用 FieldType 枚举指定转换类型
            "datetime_settings": {
                "format": "yyyy-MM-dd"
            }
        }
    }
]

# 创建类别（使用 ExtractModel 枚举）
category = client.category.create(
    workspace_id="123",
    name="发票类别",
    extract_model=ExtractModel.Acgpt,  # 使用枚举而非字符串 "Acgpt"
    sample_files=[
        "/path/to/sample1.pdf",
        "/path/to/sample2.pdf"
    ],
    fields=fields,
    category_prompt="这是发票类别"
)
print(f"类别ID: {category.category_id}")

# 一站式创建：类别 + 表格（含内嵌字段）一次完成
category = client.category.create(
    workspace_id="123",
    name="酒店水单",
    extract_model=ExtractModel.Acgpt,
    sample_files=["/path/to/sample.pdf"],
    fields=[
        {"name": "入住日期"},
        {"name": "房间号"},
    ],
    tables=[
        {
            "name": "费用明细",
            "prompt": "请抽取每行费用",
            "extract_model": "Acgpt",
            "fields": [  # 表格字段内嵌在 tables 中
                {"name": "日期"},
                {"name": "费用类型"},
                {"name": "金额"},
            ]
        }
    ],
)

获取类别列表

# 使用 EnabledStatus 枚举查询
categories = client.category.list(
    workspace_id="123",
    page=1,
    page_size=20,
    enabled=EnabledStatus.ENABLED  # 使用枚举
)

for cat in categories.categories:
    print(f"{cat.id}: {cat.name}")

字段管理

# 新增字段
field = client.category.fields.add(
    workspace_id="123",
    category_id="456",
    name="税率"
)

# 获取字段列表
fields = client.category.fields.list(
    workspace_id="123",
    category_id="456"
)

# 更新字段
client.category.fields.update(
    workspace_id="123",
    category_id="456",
    field_id="789",
    name="税率(%)",
    required=True
)

# 删除字段
client.category.fields.delete(
    workspace_id="123",
    category_id="456",
    field_ids=["789"]
)

批量操作

# 批量新增字段
result = client.category.fields.batch_add(
    workspace_id="123",
    category_id="456",
    fields=[
        {"name": "发票号码", "prompt": "发票唯一标识"},
        {"name": "金额", "description": "发票总金额"},
    ],
    with_detail=True,  # 返回完整字段详情
)
for f in result.fields:
    print(f"{f.field_id}: {f.name}")

# 批量新增表格（含内嵌字段）
result = client.category.tables.batch_add(
    workspace_id="123",
    category_id="456",
    tables=[
        {
            "name": "费用明细",
            "prompt": "抽取费用",
            "fields": [{"name": "日期"}, {"name": "金额"}],
        }
    ],
    with_detail=True,
)

# 批量上传样本
result = client.category.samples.batch_upload(
    workspace_id="123",
    category_id="456",
    files=["sample1.pdf", "sample2.pdf", "sample3.pdf"],
    with_detail=True,  # 返回全部样本列表
)

# 批量下载样本（ZIP 打包）
file_data, filename = client.category.samples.batch_download(
    workspace_id="123",
    category_id="456",
    sample_ids=["s001", "s002"],  # 不传则下载全部
)

样本管理

# 上传样本
sample = client.category.samples.upload(
    workspace_id="123",
    category_id="456",
    file="/path/to/sample.pdf"
)

# 获取样本列表
samples = client.category.samples.list(
    workspace_id="123",
    category_id="456"
)

# 下载样本
client.category.samples.download(
    workspace_id="123",
    category_id="456",
    sample_id="789",
    save_path="/path/to/save"
)

📖 更多 API 使用示例：查看 examples 目录 获取完整的使用示例，包括文件处理、审核规则和完整业务流程

4. 链式调用（减少 70% 重复代码）

通过上下文绑定，无需重复传递 workspace_id 和 category_id：

# 传统方式（冗长，需要重复传递参数）
client.category.fields.add(workspace_id="123", category_id="456", name="字段1")
client.category.fields.add(workspace_id="123", category_id="456", name="字段2")
client.category.tables.add(workspace_id="123", category_id="456", name="表格1")

# 链式调用方式（简洁，代码量减少 70%）
ws = client.workspace("123")
cat = ws.category("456")

# 简化的字段操作
cat.fields.add(name="字段1")
cat.fields.add(name="字段2")
cat.tables.add(name="表格1")

# 工作空间操作
ws.get()                      # 获取详情
ws.update(name="新名称", auth_scope=AuthScope.PUBLIC)  # 更新（name 和 auth_scope 都是必填）
ws.delete()                   # 删除

# 类别操作
cat.update(name="新名称")      # 更新类别
cat.fields.list()             # 获取字段列表
cat.tables.list()             # 获取表格列表
cat.samples.upload(file="sample.pdf")  # 上传样本

5. 自动分页迭代器

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

# 传统方式（手动分页）
all_workspaces = []
page = 1
while True:
    response = client.workspace.list(page=page, page_size=100)
    all_workspaces.extend(response.workspaces)
    if page >= response.total // response.page_size + 1:
        break
    page += 1

# 迭代器方式（自动分页）
# 方式 1: 直接遍历（内存高效，支持随时中断）
for workspace in client.workspace.iter():
    print(f"{workspace.workspace_id}: {workspace.name}")
    if some_condition:
        break  # 可以随时中断

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

# 方式 3: 限制最大页数（防止数据过大）
for workspace in client.workspace.iter(max_pages=5):
    print(workspace.name)

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

# 获取所有未启用的类别
disabled_categories = list(
    client.category.iter(workspace_id="123", enabled=EnabledStatus.DISABLED)
)

6. 枚举类型（避免参数传错）

SDK 提供完整的枚举类型定义，使用枚举可以获得 IDE 自动补全和类型检查，避免因参数拼写错误导致 API 调用失败。

6.1 可用的枚举类型

from docflow import (
    ExtractModel,      # 提取模型类型
    EnabledStatus,     # 启用状态（查询用）
    EnabledFlag,       # 启用标志（更新用）
    AuthScope,         # 权限范围
    FieldType,         # 字段类型
)

# ExtractModel - 提取模型（V1.6 命名）
ExtractModel.Auto      # 智能匹配抽取模型（字段级智能路由，由算法决定实际模型）
ExtractModel.Acgpt     # 速度快，抽取结果稳定（原 Model 1）
ExtractModel.Acgpt_VL  # 多模态，适用简单抽取（≤10 页）（原 Model 3）
ExtractModel.DF_M1     # 适用复杂文档理解（原 Model 2）
# 旧命名 ExtractModel.Model_1/Model_2/Model_3 已废弃，仍作为兼容别名保留

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

# EnabledStatus - 启用状态（查询用）
EnabledStatus.ALL       # "all" - 全部
EnabledStatus.DISABLED  # "0"   - 未启用
EnabledStatus.ENABLED   # "1"   - 已启用（默认）
EnabledStatus.OTHER     # "2"   - 其他状态

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

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

# MismatchAction - 字段值不匹配时的处理模式
MismatchAction.DEFAULT  # "default"  - 使用默认值
MismatchAction.WARNING  # "warning"  - 显示警告

6.2 使用示例

from docflow import (
    DocflowClient,
    AuthScope,
    ExtractModel,
    FieldType,
    MismatchAction,
    EnabledStatus,
    EnabledFlag,
)

client = DocflowClient.from_env()

# 使用枚举创建工作空间
workspace = client.workspace.create(
    name="我的工作空间",
    auth_scope=AuthScope.PUBLIC  # ✅ 类型安全
)

# 使用枚举创建类别
category = client.category.create(
    workspace_id="123",
    name="发票类别",
    extract_model=ExtractModel.Acgpt,  # ✅ 避免拼写错误
    sample_files=["/path/to/sample.pdf"],
    fields=[
        {
            "name": "发票号码",
            "description": "发票唯一标识"
        },
        {
            "name": "开票日期",
            "transform_settings": {
                "type": FieldType.DATETIME.value,  # ✅ 使用 FieldType 枚举
                "datetime_settings": {
                    "format": "yyyy-MM-dd"
                }
            }
        },
        {
            "name": "发票类型",
            "transform_settings": {
                "type": FieldType.ENUMERATE.value,
                "enumerate_settings": {
                    "items": ["增值税专用发票", "增值税普通发票"]
                },
                "mismatch_action": {
                    "mode": MismatchAction.WARNING.value  # ✅ 使用枚举
                }
            }
        }
    ]
)

# 使用枚举查询类别
categories = client.category.list(
    workspace_id="123",
    enabled=EnabledStatus.ENABLED  # ✅ 清晰明确
)

# 使用枚举更新类别
client.category.update(
    workspace_id="123",
    category_id="456",
    enabled=EnabledFlag.ENABLED  # ✅ 类型安全
)

# 添加字段
client.category.fields.add(
    workspace_id="123",
    category_id="456",
    name="联系电话"
)

6.3 向后兼容

所有枚举类型都向后兼容，仍然支持原有的字符串和数字参数：

# ✅ 新代码（推荐）
extract_model=ExtractModel.Acgpt
auth_scope=AuthScope.PUBLIC
transform_settings={"type": FieldType.DATETIME.value}

# ✅ 旧代码（仍然支持）
extract_model="Acgpt"          # 旧名 "Model 1" 亦兼容
auth_scope=1
transform_settings={"type": "datetime"}

7. 异常处理

from docflow.exceptions import (
    APIError,
    ValidationError,
    ResourceNotFoundError,
    PermissionDeniedError
)

try:
    workspace = client.workspace.get(workspace_id="999")
except ResourceNotFoundError as e:
    print(f"工作空间不存在: {e}")
except PermissionDeniedError as e:
    print(f"权限不足: {e}")
except APIError as e:
    print(f"API错误 [{e.status_code}]: {e.message}")

8. 使用上下文管理器

with DocflowClient(
    app_id="your-app-id",
    secret_code="your-secret-code"
) as client:
    workspaces = client.workspace.list()
    # 自动关闭连接

📖 API 文档

DocflowClient

主客户端类，用于初始化 SDK 并访问各种资源。

参数

• base_url (str): API 基础地址
• app_id (str): 应用ID（对应请求头 x-ti-app-id）
• secret_code (str): 密钥（对应请求头 x-ti-secret-code）
• timeout (int, 可选): 请求超时时间（秒），默认 30
• max_retries (int, 可选): 最大重试次数，默认 3

方法

• set_credentials(app_id: str, secret_code: str): 更新认证凭证
• close(): 关闭客户端，释放资源

WorkspaceResource

工作空间资源操作类，通过 client.workspace 访问。

方法

create()

创建工作空间

def create(
    name: str,
    auth_scope: Union[AuthScope, int],
    description: Optional[str] = None,
    **kwargs
) -> WorkspaceCreateResponse

list()

获取工作空间列表

def list(
    page: int = 1,
    page_size: int = 20
) -> WorkspaceListResponse

get()

获取工作空间详情

def get(workspace_id: str) -> WorkspaceDetailResponse

update()

更新工作空间

def update(
    workspace_id: str,
    name: str,
    auth_scope: Union[AuthScope, int],
    description: Optional[str] = None,
    callback_url: Optional[str] = None,
    callback_retry_time: Optional[int] = None,
    **kwargs
) -> None

delete()

批量删除工作空间

def delete(workspace_ids: List[str]) -> None

🔧 配置

环境变量

可以使用环境变量配置 SDK：

export DOCFLOW_BASE_URL="https://docflow.textin.com/api"
export DOCFLOW_APP_ID="your-app-id"
export DOCFLOW_SECRET_CODE="your-secret-code"

使用 .env 文件

创建 .env 文件（需要dotenv依赖）：

DOCFLOW_BASE_URL=https://docflow.textin.com/api
DOCFLOW_APP_ID=your-app-id
DOCFLOW_SECRET_CODE=your-secret-code

创建client时在代码中加载.env 配置：

from dotenv import load_dotenv
import os

load_dotenv()

client = DocflowClient(
    base_url=os.getenv("DOCFLOW_BASE_URL"),
    app_id=os.getenv("DOCFLOW_APP_ID"),
    secret_code=os.getenv("DOCFLOW_SECRET_CODE")
)

🌍 国际化（i18n）

SDK 支持多语言错误消息，方便全球化应用：

支持的语言

• zh_CN: 中文（简体）- 默认
• en_US: 英文（美国）

设置语言

方式一：创建客户端时指定

# 使用中文（默认）
client = DocflowClient(
    app_id="your-app-id",
    secret_code="your-secret-code",
    language='zh_CN'
)

# 使用英文
client = DocflowClient(
    app_id="your-app-id",
    secret_code="your-secret-code",
    language='en_US'
)

方式二：动态切换语言

client = DocflowClient(...)

# 切换到英文
client.set_language('en_US')

# 切换回中文
client.set_language('zh_CN')

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

# 获取所有可用语言
languages = client.get_available_languages()  # ['zh_CN', 'en_US']

方式三：全局设置

from docflow.i18n import set_language

# 设置全局语言（影响所有异常）
set_language('en_US')

错误消息示例

from docflow.exceptions import ValidationError, AuthenticationError

# 中文错误消息
set_language('zh_CN')
try:
    raise ValidationError(
        "字段太长",
        i18n_key='error.validation.too_long',
        field='工作空间名称',
        max_length=50
    )
except ValidationError as e:
    print(e)  # 输出: 工作空间名称 长度不能超过 50 个字符

# 英文错误消息
set_language('en_US')
try:
    raise AuthenticationError()
except AuthenticationError as e:
    print(e)  # 输出: Authentication failed, please check app_id and secret_code

🔄 重试机制

SDK 使用 urllib3.util.retry.Retry 实现智能重试功能：

自动重试的场景

以下 HTTP 状态码会自动触发重试：

• 423: Locked (资源被锁定)
• 429: Too Many Requests (请求过多，触发限流)
• 500: Internal Server Error (服务器内部错误)
• 503: Service Unavailable (服务不可用)
• 504: Gateway Timeout (网关超时)
• 900: 自定义业务错误码

重试策略

采用指数退避策略，重试间隔自动增长：

• 第 1 次重试: 等待 1 秒
• 第 2 次重试: 等待 2 秒
• 第 3 次重试: 等待 4 秒
• 第 4 次重试: 等待 8 秒

配置重试

SDK 支持灵活的重试配置，可以根据不同场景定制重试行为：

from docflow import DocflowClient

# 默认配置
client = DocflowClient(
    base_url="https://docflow.textin.com/api",
    app_id="your-app-id",
    secret_code="your-secret-code"
)

# 自定义重试次数和超时
client = DocflowClient(
    app_id="your-app-id",
    secret_code="your-secret-code",
    max_retries=5,  # 最大重试 5 次（默认 3）
    timeout=60      # 60 秒超时（默认 30）
)

# 自定义重试状态码
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",
    retry_backoff_factor=0.5,  # 更快的重试（默认 1.0）
)

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

# 查看当前重试配置
print(f"最大重试次数: {client.config.max_retries}")
print(f"重试状态码: {client.config.retry_status_codes}")
print(f"重试方法: {client.config.retry_methods}")
print(f"退避因子: {client.config.retry_backoff_factor}")

重试间隔计算公式：

等待时间 = backoff_factor * (2 ** (重试次数 - 1))

示例：

• backoff_factor=0.5: 重试间隔 0.5s, 1s, 2s, 4s...
• backoff_factor=1.0: 重试间隔 1s, 2s, 4s, 8s... （默认）
• backoff_factor=2.0: 重试间隔 2s, 4s, 8s, 16s...

不会重试的情况

以下错误不会触发重试（需要修正代码或配置）：

• 401: Unauthorized (认证失败)
• 403: Forbidden (权限不足)
• 404: Not Found (资源不存在)
• 400: Bad Request (参数错误)

🧪 运行示例

# 设置环境变量
export DOCFLOW_BASE_URL="https://docflow.textin.com/api"
export DOCFLOW_APP_ID="your-app-id"
export DOCFLOW_SECRET_CODE="your-secret-code"

# 运行快速开始示例（费用报销场景）
python examples/quick_start.py

# 运行工作流程示例（发票处理）
python examples/complete_workflow_example.py

# 运行文件处理示例
python examples/file_examples.py

# 运行审核规则示例
python examples/review_examples.py

📝 异常类型

SDK 提供以下异常类型：

• DocflowException: 基础异常类
• AuthenticationError: 认证失败异常
• ValidationError: 参数校验异常
• ResourceNotFoundError: 资源不存在异常
• PermissionDeniedError: 权限不足异常
• APIError: API 调用异常
• NetworkError: 网络异常

🔒 安全建议

1. 不要硬编码凭证：使用环境变量或配置文件管理 app_id 和 secret_code
2. 使用 HTTPS：确保 base_url 使用 HTTPS 协议
3. 定期更新凭证：使用 client.set_credentials() 定期更新认证凭证
4. 妥善保管凭证：不要将凭证提交到版本控制系统

🚀 发布新版本

发布新版本到 PyPI 的完整操作步骤：

1. 更新版本号

需要同步修改以下 3 个文件中的版本号：

# pyproject.toml
version = "x.y.z"

# docflow/__init__.py
__version__ = "x.y.z"

# setup.py
version="x.y.z",

2. 更新 CHANGELOG

在 CHANGELOG.md 顶部添加新版本的变更记录。

3. 提交并推送代码

git add -A
git commit -m "bump version to x.y.z"
git push upstream master

4. 创建并推送 tag

git tag -a vx.y.z -m "Release version x.y.z"
git push upstream vx.y.z

推送 tag 后，GitHub Actions 会自动执行：

1. 校验 tag 格式（必须匹配 v* 格式，如 v1.0.4）
2. 校验 tag 版本号与 pyproject.toml 中的版本号一致
3. 构建 Python 包
4. 发布到 PyPI（通过 trusted publisher OIDC 认证）
5. 创建 GitHub Release

注意事项

• tag 版本号（去掉 v 前缀）必须与 pyproject.toml 中的 version 完全一致，否则 CI 会报错
• PyPI 需要预先配置好 trusted publisher（仓库：intsig-textin/docflow-python-sdk，workflow：publish.yml）
• 如果 CI 失败，修复后需要删除旧 tag 并重新推送：

  git push upstream --delete vx.y.z
  git tag -d vx.y.z
  git tag -a vx.y.z -m "Release version x.y.z"
  git push upstream vx.y.z
  

🤝 贡献

欢迎提交 Issue 和 Pull Request！

📄 许可证

MIT License

---

版本: 1.0.4 最后更新: 2026-05-26