多文件上传模板上下文陷阱:当Workflow遇上多模板场景

🩺 症状

今天在Documind项目里遇到个诡异的问题:用户一次上传3个文件,分别选择Template A/B/C,Extraction阶段三个文件的Email字段都识别正常且值不同(A邮箱="alice@example.com",B邮箱="bob@test.com",C邮箱="charlie@demo.com")。但等Comparison和Data Processing跑完后,发现B和C的Email全都变成了A的值:

# Extraction结果(正常)
Template A: Email = "alice@example.com"
Template B: Email = "bob@test.com"  
Template C: Email = "charlie@demo.com"

# Comparison结果(异常)
A.Email vs B.Email:
  actual: "alice@example.com"
  expected: "alice@example.com"  ← 应该是"bob@test.com"
  
A.Email vs C.Email:
  actual: "alice@example.com"
  expected: "alice@example.com"  ← 应该是"charlie@demo.com"

Data Processing也一样,配置了Template B和C的处理规则,但运行时发现B和C的documents_by_template中对应模板没有文档,处理结果为空。

初步排查以为是前端传参错误,但检查了upload confirm API的payload,run_input_files里每个文件的template_id都正确映射了。又怀疑是Temporal workflow的多文件节点没跑完,但日志显示extraction节点per-file执行了3次,comparison和data_processing节点也正常执行了。问题到底在哪?

🔬 根因

排查了半天,最后定位到上下文传递链路断裂。代码里的坑藏在这几个地方:

upload_service.py:

# 批量上传时设置run-level template_id
first_template_id = input_files[0].get("template_id")
business_input["template_id"] = first_template_id  # ← 关键:run-level只有第一个模板
business_input["run_input_files"] = input_files  # ← 这里有完整映射

document_workflow.py (Temporal):

# Extraction节点:per-file执行(正常)
for input_file in input_files:
    await execute_extraction(input_file)  # ← 每个文件有自己的template_id
    
# Comparison节点:只执行一次(问题开始)
extraction_context = await _load_upstream_extraction_context()
# ← 从RunArtifactService读取第一条extraction payload

run_artifact_service.py:

# 离开per-file scope后,multi-file run会按id ASC解析到FIRST file row
async def get_extraction_result_by_run(run_id):
    results = await db.query(ExtractionResult).filter(run_id==run_id).order_by(id.asc()).all()
    return results[0]  # ← 默认返回第一条(Template A)

execution_service.py (Comparison):

def _resolve_field_value(extraction_context, field_name, target_template_id=None):
    # ← target_template_id传了但没用!
    return extraction_context.get(field_name)  # ← 直接从单一context取值

data_processing/service.py:

def _extraction_result_as_document(extraction_result):
    template_id = _resolve_document_template_id(extraction_result)
    # ← _resolve_document_template_id会fallback到run-level template_id
    if not template_id:
        template_id = business_input.get("template_id")  # ← fallback到first_template_id

根因一句话:批量上传时run_input_files保存了每个文件的extraction_result_id到template_id映射,但后续阶段(Comparison/Data Processing)没有用这个映射,全都fallback到run-level的first_template_id。

为什么会出现这个问题?设计时以为单文件场景够用,run-level template_id就能满足需求。多文件场景下per-file的extraction节点能正常跑,但离开per-file scope后,comparison和data_processing节点只执行一次,就"忘了"每个文件对应的模板,全都归到第一个模板上下文。

🔧 修复

修复的关键是让Comparison和Data Processing使用run_input_files的映射。具体修改:

execution_service.py (Comparison):

# ❌ 错误写法
def _resolve_field_value(extraction_context, field_name, target_template_id=None):
    return extraction_context.get(field_name)

# ✅ 正确写法
def _resolve_field_value_for_target(run_input_files, target_template_id, field_name):
    # ← 新增:根据target_template_id找到对应的extraction_result_id
    for input_file in run_input_files:
        if input_file["template_id"] == target_template_id:
            extraction_result_id = input_file["extraction_result_id"]
            # ← 用正确的extraction_result_id读取payload
            extraction_payload = await get_extraction_result_by_id(extraction_result_id)
            return extraction_payload.get(field_name)
    return None  # ← 找不到对应模板则返回None

data_processing/service.py:

# ❌ 错误写法
def _resolve_document_template_id(extraction_result):
    if not extraction_result.template_id:
        return business_input.get("template_id")  # ← fallback到run-level

# ✅ 正确写法  
def _resolve_document_template_id(extraction_result, run_input_files):
    # ← 新增参数:传入run_input_files映射
    for input_file in run_input_files:
        if input_file["extraction_result_id"] == extraction_result.id:
            return input_file["template_id"]  # ← 直接从映射取
    # ← 只有找不到时才fallback
    return business_input.get("template_id")

调用时传入run_input_files:

# Comparison调用
expected_value = _resolve_field_value_for_target(
    run_input_files,  # ← 传入完整映射
    comparison_target["templateId"],
    comparison_target["field"]
)

# Data Processing调用  
template_id = _resolve_document_template_id(
    extraction_result,
    run_input_files  # ← 传入完整映射
)

验证方法:

单元测试:构造3个文件+3个模板的run_input_files,验证Comparison能正确读取B和C的Email
集成测试:上传3个文件分别选A/B/C,检查Comparison结果中B.Email="bob@test.com",C.Email="charlie@demo.com"
数据验证:运行后检查Comparison的expected_value字段,确认与Extraction阶段的值一致

🛡️ 怎么避免

经验教训一句话:多文件场景下,不要依赖run-level的fallback,要确保上下文传递链路完整。

最佳实践:

批量操作时保存per-item映射:upload时保存extraction_result_id -> template_id,后续阶段要用这个映射
不要只在run-level保存一个值:run-level的template_id只适合单文件场景,多文件场景要用run_input_files
测试覆盖多文件+多模板场景:单元测试要覆盖"3文件3模板"的组合,不只是"1文件1模板"的单场景
代码审查时检查fallback逻辑:看到fallback到run-level时,要问"多文件场景会怎样?"

工具建议:

Temporal workflow监控:检查per-file节点是否正确传递上下文到下游节点
单元测试Mock:用Mock数据构造多文件场景,验证后续阶段读取正确的上下文
日志记录:在comparison/data_processing节点记录当前使用的template_id,方便排查

团队协作:

产品需求澄清:多文件场景的业务需求要明确"每个文件对应不同模板"
架构设计文档:多文件场景的上下文传递要写清楚,避免开发者用单文件思维写代码
Bug复盘文档:遇到类似问题要记录"上下文传递链路断裂"的排查思路

修的时候花了4小时排查,根因就2个函数没传run_input_files参数。同一个坑踩两次才叫坑,踩一次那是学费——下次遇到多文件场景,先检查上下文传递链路有没有断。